From 4bc952bc70926935d975f734a06de9cb9329953a Mon Sep 17 00:00:00 2001 From: James Long Date: Thu, 3 Sep 2026 16:05:07 +0000 Subject: [PATCH] refactor(transport): isolate local byte-stream boundary --- README.md | 20 ++ src/client.rs | 15 +- src/daemon.rs | 598 +----------------------------------------- src/daemon/server.rs | 427 ++++++++++++++++++++++++++++++ src/daemon/unix.rs | 163 ++++++++++++ src/lib.rs | 2 + src/transport/mod.rs | 16 ++ src/transport/unix.rs | 199 ++++++++++++++ tests/ownership.rs | 69 +++++ 9 files changed, 911 insertions(+), 598 deletions(-) create mode 100644 src/daemon/server.rs create mode 100644 src/daemon/unix.rs create mode 100644 src/transport/mod.rs create mode 100644 src/transport/unix.rs diff --git a/README.md b/README.md index d21aba7..4e7c1c0 100644 --- a/README.md +++ b/README.md @@ -30,6 +30,26 @@ consumes the ticket. Expiry stops an unowned daemon; if the old owner is still connected, expiry simply cancels the handoff. `shutdown` always stops the daemon, including during handoff. No ownership or handoff state is persisted. +### Local transport boundary + +`src/transport/` owns endpoint listening/connecting, byte-stream I/O, +disconnect monitoring, response completion, and cancellation. Authentication, +protocol framing, dispatch, subscriptions, and ownership stay shared in the +daemon; runtime-directory/registration wiring is platform-specific. The Unix +backend preserves half-close when completing subscriptions so queued final +frames are not discarded on macOS. Shutdown cancellation is a separate operation +that wakes partial requests and blocked subscription writes before joining. + +The planned Windows backend keeps protocol 7 and the registration fields +`instance_id`, `pid`, `protocol`, `socket`, and `token`. Treat `socket` as an opaque +local endpoint: on Windows it will be `\\.\pipe\opencode-pty-`, not a +filesystem socket. A random per-instance name, exclusive first pipe instance, +current-user access control, and rejection of remote clients protect the endpoint; +the private registration file remains the discovery and authentication source. +Named-pipe completion will use explicit bounded, cancellable delivery rather than +pretending to support Unix half-close. This boundary alone does not implement +Windows transport or daemon support. + ## Architecture ```text diff --git a/src/client.rs b/src/client.rs index 1c5515d..07ff218 100644 --- a/src/client.rs +++ b/src/client.rs @@ -12,13 +12,15 @@ use crate::protocol::{ read_subscription_event, write_frame, }; use crate::service::{CreateTerminal, TerminalId, TerminalInfo, TerminalRows}; +#[cfg(unix)] +use crate::transport::Connection; const START_TIMEOUT: Duration = Duration::from_secs(5); pub struct TerminalClient { registration: Registration, #[cfg(unix)] - owner: Option<(std::os::unix::net::UnixStream, std::process::Child)>, + owner: Option<(Connection, std::process::Child)>, } #[derive(Debug)] @@ -41,7 +43,7 @@ pub struct RemoteReplay { #[cfg(unix)] pub struct TerminalSubscription { - stream: std::os::unix::net::UnixStream, + stream: Connection, pub terminal: TerminalInfo, pub role: AttachmentRole, pub generation: u64, @@ -58,7 +60,6 @@ impl TerminalSubscription { impl TerminalClient { #[cfg(unix)] pub fn start() -> Result { - use std::os::unix::net::UnixStream; use std::os::unix::process::CommandExt; use std::process::{Command, Stdio}; @@ -82,7 +83,7 @@ impl TerminalClient { if let Ok(registration) = read_registration() && registration.pid == child.id() { - let mut stream = UnixStream::connect(®istration.socket)?; + let mut stream = Connection::connect(®istration.socket)?; stream.set_read_timeout(Some(START_TIMEOUT))?; stream.set_write_timeout(Some(START_TIMEOUT))?; write_frame( @@ -304,8 +305,7 @@ impl TerminalClient { role: AttachmentRole, takeover: bool, ) -> Result { - use std::os::unix::net::UnixStream; - let mut stream = UnixStream::connect(&self.registration.socket)?; + let mut stream = Connection::connect(&self.registration.socket)?; write_frame( &mut stream, &Envelope { @@ -362,8 +362,7 @@ impl TerminalClient { #[cfg(unix)] fn request(&self, request: Request) -> Result { - use std::os::unix::net::UnixStream; - let mut stream = UnixStream::connect(&self.registration.socket).with_context(|| { + let mut stream = Connection::connect(&self.registration.socket).with_context(|| { format!( "failed to connect to {}", self.registration.socket.display() diff --git a/src/daemon.rs b/src/daemon.rs index 219c6ac..b7aed43 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -10,603 +10,21 @@ pub struct Registration { pub instance_id: String, pub pid: u32, pub protocol: u32, + /// Opaque local transport endpoint, not necessarily a filesystem entry. pub socket: PathBuf, pub token: String, } #[cfg(unix)] -mod unix { - use std::fs::{self, OpenOptions}; - use std::net::Shutdown; - use std::os::unix::ffi::OsStrExt; - use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt}; - use std::os::unix::net::{UnixListener, UnixStream}; - use std::path::PathBuf; - use std::sync::atomic::{AtomicBool, Ordering}; - use std::sync::{Arc, Mutex}; - use std::thread; - use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; - - use anyhow::{Context, Result, anyhow}; - use base64::Engine; - use fs2::FileExt; - use sha2::{Digest, Sha256}; - - use super::{LOCK_FILE, REGISTRATION_FILE, Registration}; - use crate::ownership::Ownership; - use crate::protocol::{ - Envelope, PROTOCOL_VERSION, Request, Response, read_frame, write_frame, write_output_frame, - }; - use crate::service::{CreateTerminal, StreamEvent, TerminalService}; - - pub fn service_dir() -> PathBuf { - if let Some(path) = std::env::var_os("OPENCODE_PTY_RUNTIME_DIR") { - return PathBuf::from(path); - } - if let Some(path) = std::env::var_os("XDG_RUNTIME_DIR") { - return PathBuf::from(path).join("opencode-pty"); - } - let uid = nix::unistd::Uid::effective().as_raw(); - std::env::temp_dir().join(format!("opencode-pty-{uid}")) - } - - pub fn registration_path() -> PathBuf { - service_dir().join(REGISTRATION_FILE) - } - - pub fn read_registration() -> Result { - let data = - fs::read(registration_path()).context("opencode-pty registration is unavailable")?; - serde_json::from_slice(&data).context("invalid opencode-pty registration") - } - - pub fn run() -> Result<()> { - let directory = service_dir(); - fs::create_dir_all(&directory)?; - fs::set_permissions(&directory, fs::Permissions::from_mode(0o700))?; - let lock_path = directory.join(LOCK_FILE); - let lock = OpenOptions::new() - .create(true) - .truncate(false) - .read(true) - .write(true) - .open(&lock_path)?; - lock.try_lock_exclusive() - .context("another opencode-pty process already owns the service lock")?; - - let socket_path = socket_path(&directory)?; - if socket_path.exists() { - fs::remove_file(&socket_path)?; - } - let listener = UnixListener::bind(&socket_path)?; - fs::set_permissions(&socket_path, fs::Permissions::from_mode(0o600))?; - listener.set_nonblocking(true)?; - - let registration = Registration { - instance_id: random_id(), - pid: std::process::id(), - protocol: PROTOCOL_VERSION, - socket: socket_path.clone(), - token: random_id(), - }; - let ownership = Arc::new(Mutex::new(Ownership::new(Instant::now()))); - write_registration(&directory, ®istration)?; - - let service = Arc::new(TerminalService::default()); - let shutdown = Arc::new(AtomicBool::new(false)); - let mut handlers = Vec::<(UnixStream, thread::JoinHandle<()>)>::new(); - while !shutdown.load(Ordering::Acquire) { - if ownership - .lock() - .map_err(|_| anyhow!("ownership lock poisoned"))? - .tick(Instant::now()) - { - shutdown.store(true, Ordering::Release); - break; - } - for (_, handler) in handlers.extract_if(.., |(_, handler)| handler.is_finished()) { - let _ = handler.join(); - } - match listener.accept() { - Ok((stream, _)) => { - // macOS inherits the listener's nonblocking mode on accepted sockets. - stream.set_nonblocking(false)?; - let control = stream.try_clone()?; - let service = Arc::clone(&service); - let shutdown = Arc::clone(&shutdown); - let ownership = Arc::clone(&ownership); - let registration = registration.clone(); - let handle = thread::spawn(move || { - if let Err(error) = handle_connection( - stream, - &service, - ®istration, - &shutdown, - &ownership, - ) { - eprintln!("opencode-pty request failed: {error:#}"); - } - }); - handlers.push((control, handle)); - } - Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { - thread::sleep(Duration::from_millis(10)); - } - Err(error) => return Err(error.into()), - } - } - - drop(listener); - // Unblock partial requests, owner reads, and backpressured subscriptions - // before joining. PTY workers still use their existing termination path. - for (stream, _) in &handlers { - let _ = stream.shutdown(Shutdown::Both); - } - let (cleanup_tx, cleanup_rx) = crossbeam_channel::bounded::<()>(1); - let cleanup_registration = registration.clone(); - let watchdog = thread::spawn(move || { - if cleanup_rx.recv_timeout(Duration::from_secs(5)).is_err() { - eprintln!("opencode-pty cleanup timed out; forcing exit"); - let _ = remove_if_current(&cleanup_registration); - let _ = fs::remove_file(&cleanup_registration.socket); - std::process::exit(1); - } - }); - service.shutdown(); - for (_, handler) in handlers { - let _ = handler.join(); - } - drop(service); - remove_if_current(®istration)?; - let _ = fs::remove_file(&socket_path); - let _ = cleanup_tx.send(()); - let _ = watchdog.join(); - drop(lock); - Ok(()) - } - - fn socket_path(directory: &std::path::Path) -> Result { - let directory = - fs::canonicalize(directory).context("failed to resolve PTY runtime directory")?; - let digest = Sha256::digest(directory.as_os_str().as_bytes()); - let name = digest[..16] - .iter() - .map(|byte| format!("{byte:02x}")) - .collect::(); - let root = PathBuf::from("/tmp").join(format!( - "opencode-pty-{}", - nix::unistd::Uid::effective().as_raw() - )); - ensure_private_directory(&root)?; - Ok(root.join(format!("{name}.sock"))) - } - - fn ensure_private_directory(directory: &std::path::Path) -> Result<()> { - fs::create_dir_all(directory)?; - let metadata = fs::symlink_metadata(directory)?; - if !metadata.is_dir() || metadata.file_type().is_symlink() { - return Err(anyhow!( - "PTY socket directory is not a real directory: {}", - directory.display() - )); - } - let uid = nix::unistd::Uid::effective().as_raw(); - if metadata.uid() != uid { - return Err(anyhow!( - "PTY socket directory {} is owned by uid {}, expected {uid}", - directory.display(), - metadata.uid() - )); - } - fs::set_permissions(directory, fs::Permissions::from_mode(0o700))?; - Ok(()) - } - - fn handle_connection( - mut stream: UnixStream, - service: &TerminalService, - registration: &Registration, - shutdown: &AtomicBool, - ownership: &Mutex, - ) -> Result<()> { - let envelope: Envelope = read_frame(&mut stream)?; - if envelope.token != registration.token { - return write_frame( - &mut stream, - &Response::Error { - message: "authentication failed".to_string(), - }, - ); - } - if let Request::Own { - instance_id, - ticket, - } = envelope.request - { - let claim = if instance_id != registration.instance_id { - Err(anyhow!("daemon instance_id mismatch")) - } else if shutdown.load(Ordering::Acquire) { - Err(anyhow!("daemon is stopping")) - } else { - ownership - .lock() - .map_err(|_| anyhow!("ownership lock poisoned"))? - .claim(ticket.as_deref(), Instant::now()) - }; - if let Err(error) = claim { - return write_frame( - &mut stream, - &Response::Error { - message: error.to_string(), - }, - ); - } - let result = owner_connection(&mut stream, registration, shutdown, ownership); - ownership - .lock() - .map_err(|_| anyhow!("ownership lock poisoned"))? - .disconnect(Instant::now()); - return result; - } - if let Request::Subscribe { - id, - offset, - attachment_id, - role, - takeover, - } = envelope.request - { - return stream_subscription( - &mut stream, - service, - shutdown, - SubscriptionRequest { - id, - offset, - attachment_id, - role, - takeover, - }, - ); - } - let stopping = matches!(envelope.request, Request::Shutdown); - let response = dispatch(envelope.request, service, registration).unwrap_or_else(|error| { - Response::Error { - message: format!("{error:#}"), - } - }); - let result = write_frame(&mut stream, &response); - if stopping { - shutdown.store(true, Ordering::Release); - } - result - } - - fn owner_connection( - stream: &mut UnixStream, - registration: &Registration, - shutdown: &AtomicBool, - ownership: &Mutex, - ) -> Result<()> { - write_frame(&mut *stream, &Response::Owned)?; - while !shutdown.load(Ordering::Acquire) { - let envelope: Envelope = read_frame(&mut *stream)?; - let stopping = envelope.token == registration.token - && matches!(envelope.request, Request::Shutdown); - let response = if envelope.token != registration.token { - Response::Error { - message: "authentication failed".to_string(), - } - } else { - match envelope.request { - Request::PrepareHandoff => { - let handoff = ownership - .lock() - .map_err(|_| anyhow!("ownership lock poisoned"))? - .prepare( - Instant::now(), - SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() as u64, - )?; - Response::Handoff { - ticket: handoff.ticket, - expires_at: handoff.expires_at, - } - } - Request::Shutdown => Response::Ok, - _ => Response::Error { - message: "owner connection only accepts prepare_handoff or shutdown" - .to_string(), - }, - } - }; - let result = write_frame(&mut *stream, &response); - if stopping { - shutdown.store(true, Ordering::Release); - return result; - } - result?; - } - Ok(()) - } - - struct SubscriptionRequest { - id: crate::service::TerminalId, - offset: u64, - attachment_id: String, - role: crate::protocol::AttachmentRole, - takeover: bool, - } - - fn stream_subscription( - stream: &mut UnixStream, - service: &TerminalService, - shutdown: &AtomicBool, - request: SubscriptionRequest, - ) -> Result<()> { - use std::io::Read; - use std::net::Shutdown; - - let attachment = service.attach( - request.id, - request.offset, - request.attachment_id, - request.role, - request.takeover, - )?; - write_frame( - &mut *stream, - &Response::Attached { - terminal: attachment.terminal.clone(), - role: attachment.role, - generation: attachment.generation, - requested_offset: attachment.replay.requested_offset, - available_offset: attachment.replay.available_offset, - end_offset: attachment.replay.end_offset, - truncated: attachment.replay.truncated, - replay_base64: base64::engine::general_purpose::STANDARD - .encode(&attachment.replay.bytes), - }, - )?; - let mut monitor_stream = stream.try_clone()?; - let (disconnect_tx, disconnect_rx) = crossbeam_channel::bounded::<()>(1); - let monitor = thread::spawn(move || { - let mut byte = [0_u8; 1]; - let _ = monitor_stream.read(&mut byte); - let _ = disconnect_tx.send(()); - }); - let result = (|| loop { - if shutdown.load(Ordering::Acquire) { - break Ok(()); - } - let event = crossbeam_channel::select! { - recv(disconnect_rx) -> _ => break Ok(()), - recv(attachment.events) -> event => match event { - Ok(event) => event, - Err(_) => break Ok(()), - }, - default(Duration::from_millis(100)) => continue, - }; - let response = match event { - StreamEvent::Output { start, end, bytes } => { - write_output_frame(&mut *stream, start, end, &bytes)?; - continue; - } - StreamEvent::Resized { - cols, - rows, - generation, - checkpoint, - } => Response::Resized { - cols, - rows, - generation, - checkpoint_base64: base64::engine::general_purpose::STANDARD.encode(checkpoint), - }, - StreamEvent::Exited { - exit_code, - final_offset, - } => Response::Exited { - exit_code, - final_offset, - }, - StreamEvent::ControllerChanged { - attachment_id, - generation, - } => Response::ControllerChanged { - attachment_id, - generation, - }, - StreamEvent::TitleChanged { title } => Response::TitleChanged { title }, - StreamEvent::ForegroundProcessChanged { process } => { - Response::ForegroundProcessChanged { process } - } - }; - write_frame(&mut *stream, &response)?; - if matches!(response, Response::Exited { .. }) { - break Ok(()); - } - })(); - // A full shutdown can discard a just-written final frame on macOS. - // Half-close first so the peer drains queued output before closing. - let _ = stream.shutdown(Shutdown::Write); - let _ = monitor.join(); - result - } - - fn dispatch( - request: Request, - service: &TerminalService, - registration: &Registration, - ) -> Result { - Ok(match request { - Request::Ping => Response::Pong { - instance_id: registration.instance_id.clone(), - pid: registration.pid, - protocol: registration.protocol, - }, - Request::Create { - program, - args, - cwd, - title, - group_id, - env, - cols, - rows, - } => Response::Created { - terminal: service.create(CreateTerminal { - program, - args, - cwd, - title, - group_id, - env, - cols, - rows, - })?, - }, - Request::List => Response::Terminals { - terminals: service.list()?, - }, - Request::Write { - id, - attachment_id, - data_base64, - } => { - let bytes = base64::engine::general_purpose::STANDARD - .decode(data_base64) - .context("invalid input base64")?; - service.write_for(id, attachment_id, bytes)?; - Response::Ok - } - Request::Resize { - id, - attachment_id, - cols, - rows, - } => { - service.resize_for(id, attachment_id, cols, rows)?; - Response::Ok - } - Request::Control { - id, - attachment_id, - cols, - rows, - } => { - service.control(id, attachment_id, cols, rows)?; - Response::Ok - } - Request::Input { - id, - attachment_id, - cols, - rows, - data_base64, - } => { - let bytes = base64::engine::general_purpose::STANDARD - .decode(data_base64) - .context("invalid input base64")?; - service.input(id, attachment_id, cols, rows, bytes)?; - Response::Ok - } - Request::Snapshot { id } => { - let snapshot = service.snapshot(id)?; - Response::Snapshot { - terminal: snapshot.info, - text: snapshot.text, - checkpoint_base64: base64::engine::general_purpose::STANDARD - .encode(snapshot.checkpoint), - cursor_x: snapshot.cursor_x, - cursor_y: snapshot.cursor_y, - } - } - Request::ReadRows { id, rows } => { - let rows = service.read_rows(id, rows)?; - Response::Rows { - terminal: rows.terminal, - lines: rows.lines, - cursor_x: rows.cursor_x, - cursor_y: rows.cursor_y, - } - } - Request::Replay { id, offset } => { - let replay = service.replay(id, offset)?; - Response::Replay { - requested_offset: replay.requested_offset, - available_offset: replay.available_offset, - end_offset: replay.end_offset, - truncated: replay.truncated, - data_base64: base64::engine::general_purpose::STANDARD.encode(replay.bytes), - } - } - Request::Subscribe { .. } => unreachable!("subscriptions are handled before dispatch"), - Request::Terminate { id } => { - service.terminate(id)?; - Response::Ok - } - Request::Shutdown => Response::Ok, - Request::Own { .. } => unreachable!("ownership is handled before dispatch"), - Request::PrepareHandoff => Response::Error { - message: "handoff requires the owner connection".to_string(), - }, - }) - } - - fn write_registration(directory: &std::path::Path, registration: &Registration) -> Result<()> { - let temporary = directory.join(format!("service.{}.tmp", registration.instance_id)); - let data = serde_json::to_vec_pretty(registration)?; - let mut file = OpenOptions::new() - .create_new(true) - .write(true) - .mode(0o600) - .open(&temporary)?; - use std::io::Write; - file.write_all(&data)?; - file.sync_all()?; - fs::rename(&temporary, registration_path())?; - Ok(()) - } - - fn remove_if_current(registration: &Registration) -> Result<()> { - if read_registration().is_ok_and(|current| current.instance_id == registration.instance_id) - { - fs::remove_file(registration_path())?; - } - Ok(()) - } - - fn random_id() -> String { - format!("{:032x}", rand::random::()) - } - - #[cfg(test)] - mod tests { - use super::*; - - #[test] - fn socket_paths_are_short_and_runtime_specific() { - let base = - std::env::temp_dir().join(format!("opencode-pty-socket-test-{}", random_id())); - let first = base.join("a".repeat(120)).join("database-a"); - let second = base.join("b".repeat(120)).join("database-b"); - fs::create_dir_all(&first).unwrap(); - fs::create_dir_all(&second).unwrap(); - - let first_socket = socket_path(&first).unwrap(); - let second_socket = socket_path(&second).unwrap(); - - assert_ne!(first_socket, second_socket); - assert!(first_socket.as_os_str().as_bytes().len() < 104); - assert_eq!(first_socket.parent(), second_socket.parent()); - - fs::remove_dir_all(base).unwrap(); - } - } -} +#[path = "daemon/unix.rs"] +mod platform; +#[cfg(unix)] +mod server; #[cfg(unix)] -pub use unix::{read_registration, registration_path, run, service_dir}; +pub use platform::{read_registration, registration_path, service_dir}; +#[cfg(unix)] +pub use server::run; #[cfg(not(unix))] pub fn run() -> anyhow::Result<()> { diff --git a/src/daemon/server.rs b/src/daemon/server.rs new file mode 100644 index 0000000..3b23ae8 --- /dev/null +++ b/src/daemon/server.rs @@ -0,0 +1,427 @@ +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use anyhow::{Context, Result, anyhow}; +use base64::Engine; + +use super::{Registration, platform}; +use crate::ownership::Ownership; +use crate::protocol::{Envelope, Request, Response, read_frame, write_frame, write_output_frame}; +use crate::service::{CreateTerminal, StreamEvent, TerminalService}; +use crate::transport::{Cancellation, Connection}; + +pub fn run() -> Result<()> { + let (runtime, mut listener) = platform::Runtime::bind()?; + let registration = &runtime.registration; + let ownership = Arc::new(Mutex::new(Ownership::new(Instant::now()))); + + let service = Arc::new(TerminalService::default()); + let shutdown = Arc::new(AtomicBool::new(false)); + let mut handlers = Vec::<(Cancellation, thread::JoinHandle<()>)>::new(); + while !shutdown.load(Ordering::Acquire) { + if ownership + .lock() + .map_err(|_| anyhow!("ownership lock poisoned"))? + .tick(Instant::now()) + { + shutdown.store(true, Ordering::Release); + break; + } + for (_, handler) in handlers.extract_if(.., |(_, handler)| handler.is_finished()) { + let _ = handler.join(); + } + match listener.accept() { + Ok(Some(stream)) => { + let control = stream.cancellation()?; + let service = Arc::clone(&service); + let shutdown = Arc::clone(&shutdown); + let ownership = Arc::clone(&ownership); + let registration = registration.clone(); + let handle = thread::spawn(move || { + if let Err(error) = + handle_connection(stream, &service, ®istration, &shutdown, &ownership) + { + eprintln!("opencode-pty request failed: {error:#}"); + } + }); + handlers.push((control, handle)); + } + Ok(None) => { + thread::sleep(Duration::from_millis(10)); + } + Err(error) => return Err(error.into()), + } + } + + drop(listener); + // Unblock partial requests, owner reads, and backpressured subscriptions + // before joining. PTY workers still use their existing termination path. + for (cancellation, _) in &handlers { + cancellation.cancel(); + } + let (cleanup_tx, cleanup_rx) = crossbeam_channel::bounded::<()>(1); + let cleanup_registration = registration.clone(); + let watchdog = thread::spawn(move || { + if cleanup_rx.recv_timeout(Duration::from_secs(5)).is_err() { + eprintln!("opencode-pty cleanup timed out; forcing exit"); + let _ = platform::cleanup(&cleanup_registration); + std::process::exit(1); + } + }); + service.shutdown(); + for (_, handler) in handlers { + let _ = handler.join(); + } + drop(service); + platform::cleanup(registration)?; + let _ = cleanup_tx.send(()); + let _ = watchdog.join(); + drop(runtime); + Ok(()) +} + +fn handle_connection( + mut stream: Connection, + service: &TerminalService, + registration: &Registration, + shutdown: &AtomicBool, + ownership: &Mutex, +) -> Result<()> { + let envelope: Envelope = read_frame(&mut stream)?; + if envelope.token != registration.token { + return send_response( + &mut stream, + &Response::Error { + message: "authentication failed".to_string(), + }, + ); + } + if let Request::Own { + instance_id, + ticket, + } = envelope.request + { + let claim = if instance_id != registration.instance_id { + Err(anyhow!("daemon instance_id mismatch")) + } else if shutdown.load(Ordering::Acquire) { + Err(anyhow!("daemon is stopping")) + } else { + ownership + .lock() + .map_err(|_| anyhow!("ownership lock poisoned"))? + .claim(ticket.as_deref(), Instant::now()) + }; + if let Err(error) = claim { + return send_response( + &mut stream, + &Response::Error { + message: error.to_string(), + }, + ); + } + let result = owner_connection(&mut stream, registration, shutdown, ownership); + ownership + .lock() + .map_err(|_| anyhow!("ownership lock poisoned"))? + .disconnect(Instant::now()); + return result; + } + if let Request::Subscribe { + id, + offset, + attachment_id, + role, + takeover, + } = envelope.request + { + return stream_subscription( + &mut stream, + service, + shutdown, + SubscriptionRequest { + id, + offset, + attachment_id, + role, + takeover, + }, + ); + } + let stopping = matches!(envelope.request, Request::Shutdown); + let response = + dispatch(envelope.request, service, registration).unwrap_or_else(|error| Response::Error { + message: format!("{error:#}"), + }); + let result = send_response(&mut stream, &response); + if stopping { + shutdown.store(true, Ordering::Release); + } + result +} + +fn owner_connection( + stream: &mut Connection, + registration: &Registration, + shutdown: &AtomicBool, + ownership: &Mutex, +) -> Result<()> { + write_frame(&mut *stream, &Response::Owned)?; + while !shutdown.load(Ordering::Acquire) { + let envelope: Envelope = read_frame(&mut *stream)?; + let stopping = + envelope.token == registration.token && matches!(envelope.request, Request::Shutdown); + let response = if envelope.token != registration.token { + Response::Error { + message: "authentication failed".to_string(), + } + } else { + match envelope.request { + Request::PrepareHandoff => { + let handoff = ownership + .lock() + .map_err(|_| anyhow!("ownership lock poisoned"))? + .prepare( + Instant::now(), + SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() as u64, + )?; + Response::Handoff { + ticket: handoff.ticket, + expires_at: handoff.expires_at, + } + } + Request::Shutdown => Response::Ok, + _ => Response::Error { + message: "owner connection only accepts prepare_handoff or shutdown" + .to_string(), + }, + } + }; + if stopping { + let result = send_response(stream, &response); + shutdown.store(true, Ordering::Release); + return result; + } + write_frame(&mut *stream, &response)?; + } + Ok(()) +} + +struct SubscriptionRequest { + id: crate::service::TerminalId, + offset: u64, + attachment_id: String, + role: crate::protocol::AttachmentRole, + takeover: bool, +} + +fn stream_subscription( + stream: &mut Connection, + service: &TerminalService, + shutdown: &AtomicBool, + request: SubscriptionRequest, +) -> Result<()> { + let attachment = service.attach( + request.id, + request.offset, + request.attachment_id, + request.role, + request.takeover, + )?; + write_frame( + &mut *stream, + &Response::Attached { + terminal: attachment.terminal.clone(), + role: attachment.role, + generation: attachment.generation, + requested_offset: attachment.replay.requested_offset, + available_offset: attachment.replay.available_offset, + end_offset: attachment.replay.end_offset, + truncated: attachment.replay.truncated, + replay_base64: base64::engine::general_purpose::STANDARD + .encode(&attachment.replay.bytes), + }, + )?; + let monitor = stream.monitor_disconnect()?; + let result = (|| loop { + if shutdown.load(Ordering::Acquire) { + break Ok(()); + } + let event = crossbeam_channel::select! { + recv(monitor.disconnected()) -> _ => break Ok(()), + recv(attachment.events) -> event => match event { + Ok(event) => event, + Err(_) => break Ok(()), + }, + default(Duration::from_millis(100)) => continue, + }; + let response = match event { + StreamEvent::Output { start, end, bytes } => { + write_output_frame(&mut *stream, start, end, &bytes)?; + continue; + } + StreamEvent::Resized { + cols, + rows, + generation, + checkpoint, + } => Response::Resized { + cols, + rows, + generation, + checkpoint_base64: base64::engine::general_purpose::STANDARD.encode(checkpoint), + }, + StreamEvent::Exited { + exit_code, + final_offset, + } => Response::Exited { + exit_code, + final_offset, + }, + StreamEvent::ControllerChanged { + attachment_id, + generation, + } => Response::ControllerChanged { + attachment_id, + generation, + }, + StreamEvent::TitleChanged { title } => Response::TitleChanged { title }, + StreamEvent::ForegroundProcessChanged { process } => { + Response::ForegroundProcessChanged { process } + } + }; + write_frame(&mut *stream, &response)?; + if matches!(response, Response::Exited { .. }) { + break Ok(()); + } + })(); + monitor.finish(); + result +} + +fn send_response(stream: &mut Connection, response: &Response) -> Result<()> { + write_frame(&mut *stream, response)?; + stream.finish_response()?; + Ok(()) +} + +fn dispatch( + request: Request, + service: &TerminalService, + registration: &Registration, +) -> Result { + Ok(match request { + Request::Ping => Response::Pong { + instance_id: registration.instance_id.clone(), + pid: registration.pid, + protocol: registration.protocol, + }, + Request::Create { + program, + args, + cwd, + title, + group_id, + env, + cols, + rows, + } => Response::Created { + terminal: service.create(CreateTerminal { + program, + args, + cwd, + title, + group_id, + env, + cols, + rows, + })?, + }, + Request::List => Response::Terminals { + terminals: service.list()?, + }, + Request::Write { + id, + attachment_id, + data_base64, + } => { + let bytes = base64::engine::general_purpose::STANDARD + .decode(data_base64) + .context("invalid input base64")?; + service.write_for(id, attachment_id, bytes)?; + Response::Ok + } + Request::Resize { + id, + attachment_id, + cols, + rows, + } => { + service.resize_for(id, attachment_id, cols, rows)?; + Response::Ok + } + Request::Control { + id, + attachment_id, + cols, + rows, + } => { + service.control(id, attachment_id, cols, rows)?; + Response::Ok + } + Request::Input { + id, + attachment_id, + cols, + rows, + data_base64, + } => { + let bytes = base64::engine::general_purpose::STANDARD + .decode(data_base64) + .context("invalid input base64")?; + service.input(id, attachment_id, cols, rows, bytes)?; + Response::Ok + } + Request::Snapshot { id } => { + let snapshot = service.snapshot(id)?; + Response::Snapshot { + terminal: snapshot.info, + text: snapshot.text, + checkpoint_base64: base64::engine::general_purpose::STANDARD + .encode(snapshot.checkpoint), + cursor_x: snapshot.cursor_x, + cursor_y: snapshot.cursor_y, + } + } + Request::ReadRows { id, rows } => { + let rows = service.read_rows(id, rows)?; + Response::Rows { + terminal: rows.terminal, + lines: rows.lines, + cursor_x: rows.cursor_x, + cursor_y: rows.cursor_y, + } + } + Request::Replay { id, offset } => { + let replay = service.replay(id, offset)?; + Response::Replay { + requested_offset: replay.requested_offset, + available_offset: replay.available_offset, + end_offset: replay.end_offset, + truncated: replay.truncated, + data_base64: base64::engine::general_purpose::STANDARD.encode(replay.bytes), + } + } + Request::Subscribe { .. } => unreachable!("subscriptions are handled before dispatch"), + Request::Terminate { id } => { + service.terminate(id)?; + Response::Ok + } + Request::Shutdown => Response::Ok, + Request::Own { .. } => unreachable!("ownership is handled before dispatch"), + Request::PrepareHandoff => Response::Error { + message: "handoff requires the owner connection".to_string(), + }, + }) +} diff --git a/src/daemon/unix.rs b/src/daemon/unix.rs new file mode 100644 index 0000000..c353ae3 --- /dev/null +++ b/src/daemon/unix.rs @@ -0,0 +1,163 @@ +use std::fs::{self, File, OpenOptions}; +use std::io::Write; +use std::os::unix::ffi::OsStrExt; +use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt}; +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result, anyhow}; +use fs2::FileExt; +use sha2::{Digest, Sha256}; + +use super::{LOCK_FILE, REGISTRATION_FILE, Registration}; +use crate::protocol::PROTOCOL_VERSION; +use crate::transport::Listener; + +pub fn service_dir() -> PathBuf { + if let Some(path) = std::env::var_os("OPENCODE_PTY_RUNTIME_DIR") { + return PathBuf::from(path); + } + if let Some(path) = std::env::var_os("XDG_RUNTIME_DIR") { + return PathBuf::from(path).join("opencode-pty"); + } + let uid = nix::unistd::Uid::effective().as_raw(); + std::env::temp_dir().join(format!("opencode-pty-{uid}")) +} + +pub fn registration_path() -> PathBuf { + service_dir().join(REGISTRATION_FILE) +} + +pub fn read_registration() -> Result { + let data = fs::read(registration_path()).context("opencode-pty registration is unavailable")?; + serde_json::from_slice(&data).context("invalid opencode-pty registration") +} + +pub(super) struct Runtime { + pub registration: Registration, + // Held until the shared server has joined its handlers and removed registration. + _lock: File, +} + +impl Runtime { + pub fn bind() -> Result<(Self, Listener)> { + let directory = service_dir(); + fs::create_dir_all(&directory)?; + fs::set_permissions(&directory, fs::Permissions::from_mode(0o700))?; + let lock = OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(directory.join(LOCK_FILE))?; + lock.try_lock_exclusive() + .context("another opencode-pty process already owns the service lock")?; + + let socket = socket_path(&directory)?; + if socket.exists() { + fs::remove_file(&socket)?; + } + let listener = Listener::bind(&socket)?; + fs::set_permissions(&socket, fs::Permissions::from_mode(0o600))?; + let registration = Registration { + instance_id: random_id(), + pid: std::process::id(), + protocol: PROTOCOL_VERSION, + socket, + token: random_id(), + }; + write_registration(&directory, ®istration)?; + Ok(( + Self { + registration, + _lock: lock, + }, + listener, + )) + } +} + +fn socket_path(directory: &Path) -> Result { + let directory = + fs::canonicalize(directory).context("failed to resolve PTY runtime directory")?; + let digest = Sha256::digest(directory.as_os_str().as_bytes()); + let name = digest[..16] + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + let root = PathBuf::from("/tmp").join(format!( + "opencode-pty-{}", + nix::unistd::Uid::effective().as_raw() + )); + ensure_private_directory(&root)?; + Ok(root.join(format!("{name}.sock"))) +} + +fn ensure_private_directory(directory: &Path) -> Result<()> { + fs::create_dir_all(directory)?; + let metadata = fs::symlink_metadata(directory)?; + if !metadata.is_dir() || metadata.file_type().is_symlink() { + return Err(anyhow!( + "PTY socket directory is not a real directory: {}", + directory.display() + )); + } + let uid = nix::unistd::Uid::effective().as_raw(); + if metadata.uid() != uid { + return Err(anyhow!( + "PTY socket directory {} is owned by uid {}, expected {uid}", + directory.display(), + metadata.uid() + )); + } + fs::set_permissions(directory, fs::Permissions::from_mode(0o700))?; + Ok(()) +} + +fn write_registration(directory: &Path, registration: &Registration) -> Result<()> { + let temporary = directory.join(format!("service.{}.tmp", registration.instance_id)); + let data = serde_json::to_vec_pretty(registration)?; + let mut file = OpenOptions::new() + .create_new(true) + .write(true) + .mode(0o600) + .open(&temporary)?; + file.write_all(&data)?; + file.sync_all()?; + fs::rename(&temporary, registration_path())?; + Ok(()) +} + +pub(super) fn cleanup(registration: &Registration) -> Result<()> { + if read_registration().is_ok_and(|current| current.instance_id == registration.instance_id) { + fs::remove_file(registration_path())?; + } + let _ = fs::remove_file(®istration.socket); + Ok(()) +} + +fn random_id() -> String { + format!("{:032x}", rand::random::()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn socket_paths_are_short_and_runtime_specific() { + let base = std::env::temp_dir().join(format!("opencode-pty-socket-test-{}", random_id())); + let first = base.join("a".repeat(120)).join("database-a"); + let second = base.join("b".repeat(120)).join("database-b"); + fs::create_dir_all(&first).unwrap(); + fs::create_dir_all(&second).unwrap(); + + let first_socket = socket_path(&first).unwrap(); + let second_socket = socket_path(&second).unwrap(); + + assert_ne!(first_socket, second_socket); + assert!(first_socket.as_os_str().as_bytes().len() < 104); + assert_eq!(first_socket.parent(), second_socket.parent()); + + fs::remove_dir_all(base).unwrap(); + } +} diff --git a/src/lib.rs b/src/lib.rs index 0a435f8..0eba3b7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,3 +5,5 @@ mod ghostty; mod ownership; pub mod protocol; pub mod service; +#[cfg(unix)] +mod transport; diff --git a/src/transport/mod.rs b/src/transport/mod.rs new file mode 100644 index 0000000..1e165e3 --- /dev/null +++ b/src/transport/mod.rs @@ -0,0 +1,16 @@ +//! Local byte-stream transport boundary. Protocol framing and authentication live +//! above this module. Cancellation interrupts pending I/O; successful response +//! completion is separate so cancelling a connection cannot discard its final +//! frame before the peer has had a chance to read it. +//! +//! The daemon polls `Listener::accept`; connection reads/writes run only on its +//! handler threads. A separate `Cancellation` can wake both directions without +//! waiting for either handler or peer. `finish_response` and a disconnect +//! monitor's `finish` are delivery operations, not aliases for cancellation. +//! In particular, a future named-pipe backend must not emulate Unix half-close: +//! it must retain final bytes until peer close or a bounded, cancellable deadline. + +#[cfg(unix)] +mod unix; +#[cfg(unix)] +pub(crate) use unix::{Cancellation, Connection, Listener}; diff --git a/src/transport/unix.rs b/src/transport/unix.rs new file mode 100644 index 0000000..b2c5ccd --- /dev/null +++ b/src/transport/unix.rs @@ -0,0 +1,199 @@ +use std::io::{self, Read, Write}; +use std::net::Shutdown; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::Path; +use std::thread::{self, JoinHandle}; +use std::time::Duration; + +use crossbeam_channel::{Receiver, bounded}; + +pub(crate) struct Listener(UnixListener); + +impl Listener { + pub fn bind(endpoint: &Path) -> io::Result { + let listener = UnixListener::bind(endpoint)?; + listener.set_nonblocking(true)?; + Ok(Self(listener)) + } + + /// Poll for a connection without blocking the daemon control path. + pub fn accept(&mut self) -> io::Result> { + match self.0.accept() { + Ok((stream, _)) => { + // macOS inherits the listener's nonblocking mode on accepted sockets. + stream.set_nonblocking(false)?; + Ok(Some(Connection(stream))) + } + Err(error) if error.kind() == io::ErrorKind::WouldBlock => Ok(None), + Err(error) => Err(error), + } + } +} + +pub(crate) struct Connection(UnixStream); + +impl Connection { + pub fn connect(endpoint: &Path) -> io::Result { + UnixStream::connect(endpoint).map(Self) + } + + pub fn set_read_timeout(&self, timeout: Option) -> io::Result<()> { + self.0.set_read_timeout(timeout) + } + + pub fn set_write_timeout(&self, timeout: Option) -> io::Result<()> { + self.0.set_write_timeout(timeout) + } + + pub fn cancellation(&self) -> io::Result { + self.0.try_clone().map(Cancellation) + } + + /// Unix close preserves queued response bytes; no peer acknowledgement is needed. + pub fn finish_response(&mut self) -> io::Result<()> { + Ok(()) + } + + /// Subscriptions have no more request input. A read detects peer closure (or + /// unexpected extra input) while the handler writes events. + pub fn monitor_disconnect(&self) -> io::Result { + let mut reader = self.0.try_clone()?; + let stream = self.0.try_clone()?; + let (sender, disconnected) = bounded(1); + let thread = thread::spawn(move || { + let _ = reader.read(&mut [0_u8; 1]); + let _ = sender.send(()); + }); + Ok(DisconnectMonitor { + stream, + disconnected, + thread, + }) + } +} + +impl Read for Connection { + fn read(&mut self, bytes: &mut [u8]) -> io::Result { + self.0.read(bytes) + } +} + +impl Write for Connection { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.0.write(bytes) + } + + fn flush(&mut self) -> io::Result<()> { + self.0.flush() + } +} + +pub(crate) struct Cancellation(UnixStream); + +impl Cancellation { + /// Idempotently wake partial request reads and backpressured writes. + pub fn cancel(&self) { + let _ = self.0.shutdown(Shutdown::Both); + } +} + +pub(crate) struct DisconnectMonitor { + stream: UnixStream, + disconnected: Receiver<()>, + thread: JoinHandle<()>, +} + +impl DisconnectMonitor { + pub fn disconnected(&self) -> &Receiver<()> { + &self.disconnected + } + + pub fn finish(self) { + // A full shutdown can discard a just-written final frame on macOS. + // Half-close first so the peer drains queued output before closing. + // Daemon-wide cancellation still interrupts this wait during shutdown. + let _ = self.stream.shutdown(Shutdown::Write); + let _ = self.thread.join(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::protocol::{Response, read_frame, write_frame}; + + #[test] + fn listener_polls_and_connections_roundtrip() { + let endpoint = + std::env::temp_dir().join(format!("pty-{:032x}.sock", rand::random::())); + let mut listener = Listener::bind(&endpoint).unwrap(); + assert!(listener.accept().unwrap().is_none()); + let mut client = Connection::connect(&endpoint).unwrap(); + client + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut server = listener.accept().unwrap().unwrap(); + write_frame(&mut client, &Response::Owned).unwrap(); + assert!(matches!(read_frame(&mut server).unwrap(), Response::Owned)); + write_frame(&mut server, &Response::Ok).unwrap(); + server.finish_response().unwrap(); + assert!(matches!(read_frame(&mut client).unwrap(), Response::Ok)); + drop(listener); + std::fs::remove_file(endpoint).unwrap(); + } + + #[test] + fn cancellation_wakes_partial_reads_and_is_idempotent() { + let (reader, mut peer) = UnixStream::pair().unwrap(); + let mut connection = Connection(reader); + let cancellation = connection.cancellation().unwrap(); + peer.write_all(&[0, 0]).unwrap(); + let (sender, receiver) = bounded(1); + let reader = thread::spawn(move || { + sender + .send(read_frame::(&mut connection)) + .unwrap(); + }); + cancellation.cancel(); + cancellation.cancel(); + assert!( + receiver + .recv_timeout(Duration::from_secs(2)) + .unwrap() + .is_err() + ); + reader.join().unwrap(); + } + + #[test] + fn subscription_completion_preserves_final_frame() { + let (server, client) = UnixStream::pair().unwrap(); + let mut server = Connection(server); + let mut client = Connection(client); + client + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let monitor = server.monitor_disconnect().unwrap(); + let writer = thread::spawn(move || { + write_frame( + &mut server, + &Response::Exited { + exit_code: Some(0), + final_offset: 123, + }, + ) + .unwrap(); + monitor.finish(); + }); + assert!(matches!( + read_frame(&mut client).unwrap(), + Response::Exited { + exit_code: Some(0), + final_offset: 123, + } + )); + assert_eq!(client.read(&mut [0]).unwrap(), 0); + drop(client); + writer.join().unwrap(); + } +} diff --git a/tests/ownership.rs b/tests/ownership.rs index e574202..5e01d46 100644 --- a/tests/ownership.rs +++ b/tests/ownership.rs @@ -163,6 +163,21 @@ fn assert_terminal_stopped(terminal: &TerminalInfo) { #[test] fn owner_loss_stops_terminals_despite_blocked_clients() { let mut daemon = Daemon::start(); + let mut unauthenticated = daemon.connect(); + write_frame( + &mut unauthenticated, + &Envelope { + token: "wrong-token".into(), + request: Request::Own { + instance_id: daemon.registration.instance_id.clone(), + ticket: None, + }, + }, + ) + .unwrap(); + assert!( + matches!(read_frame(&mut unauthenticated).unwrap(), Response::Error { message } if message == "authentication failed") + ); assert!(matches!( daemon.request(Request::Own { instance_id: "wrong-instance".into(), @@ -197,6 +212,60 @@ fn owner_loss_stops_terminals_despite_blocked_clients() { assert_terminal_stopped(&terminal); } +#[test] +fn subscription_delivers_final_exit_frame_before_closing() { + use opencode_pty::protocol::{SubscriptionEvent, read_subscription_event}; + + let mut daemon = Daemon::start(); + let (owner, response) = daemon.own(None); + assert!(matches!(response, Response::Owned)); + let terminal = daemon.terminal("read line; printf final-output"); + let mut subscription = daemon.connect(); + assert!(matches!( + daemon.send( + &mut subscription, + Request::Subscribe { + id: terminal.id, + offset: 0, + attachment_id: "final-observer".into(), + role: AttachmentRole::Observer, + takeover: false, + } + ), + Response::Attached { .. } + )); + assert!(matches!( + daemon.request(Request::Write { + id: terminal.id, + attachment_id: None, + data_base64: base64::engine::general_purpose::STANDARD.encode(b"finish\n"), + }), + Response::Ok + )); + let mut output = Vec::new(); + loop { + match read_subscription_event(&mut subscription).unwrap() { + SubscriptionEvent::Output { bytes, .. } => output.extend(bytes), + SubscriptionEvent::Response(response) => match *response { + Response::Exited { + exit_code, + final_offset, + } => { + assert_eq!(exit_code, Some(0)); + assert_eq!(final_offset, output.len() as u64); + break; + } + Response::Error { message } => panic!("subscription error: {message}"), + _ => {} + }, + } + } + assert!(String::from_utf8_lossy(&output).contains("final-output")); + drop(subscription); + drop(owner); + daemon.wait(); +} + #[test] fn handoff_preserves_daemon_and_terminal_and_shutdown_overrides_it() { let mut daemon = Daemon::start();