diff --git a/crates/pilotty-cli/src/daemon/client.rs b/crates/pilotty-cli/src/daemon/client.rs index 8bfaa35..a364dbb 100644 --- a/crates/pilotty-cli/src/daemon/client.rs +++ b/crates/pilotty-cli/src/daemon/client.rs @@ -5,7 +5,10 @@ use std::process::Stdio; use std::time::Duration; use anyhow::{bail, Context, Result}; -use pilotty_core::protocol::{Request, Response, PROTOCOL_VERSION}; +use pilotty_core::error::ApiError; +use pilotty_core::protocol::{ + supports_protocol, Command, Request, Response, LEGACY_PROTOCOL_VERSION, PROTOCOL_VERSION, +}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; use tokio::time::timeout; @@ -22,6 +25,26 @@ const RETRY_INTERVAL: Duration = Duration::from_millis(100); /// Client for communicating with the daemon. pub struct DaemonClient { stream: UnixStream, + daemon_protocol: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ProtocolAction { + Send, + Probe, + Reject { observed: u32, required: u32 }, +} + +fn protocol_action(observed: Option, required: u32) -> ProtocolAction { + if required == LEGACY_PROTOCOL_VERSION { + return ProtocolAction::Send; + } + + match observed { + None => ProtocolAction::Probe, + Some(observed) if supports_protocol(observed, required) => ProtocolAction::Send, + Some(observed) => ProtocolAction::Reject { observed, required }, + } } impl DaemonClient { @@ -32,7 +55,10 @@ impl DaemonClient { // Try to connect directly first if let Ok(stream) = UnixStream::connect(&socket_path).await { debug!("Connected to existing daemon"); - return Ok(Self { stream }); + return Ok(Self { + stream, + daemon_protocol: None, + }); } // Daemon not running, start it @@ -41,7 +67,10 @@ impl DaemonClient { // Wait for daemon to become available, checking if it crashes let stream = Self::wait_for_daemon(&socket_path, child).await?; - Ok(Self { stream }) + Ok(Self { + stream, + daemon_protocol: None, + }) } /// Start the daemon as a background process. @@ -122,6 +151,49 @@ impl DaemonClient { request: Request, timeout_duration: Duration, ) -> Result { + let required = request.command.minimum_protocol(); + self.ensure_protocol(&request.id, required, timeout_duration) + .await?; + + let response = self.exchange(request, timeout_duration).await?; + self.daemon_protocol = Some(response.protocol); + + if response.protocol < PROTOCOL_VERSION { + eprintln!( + "warning: the running daemon speaks protocol {} but this client speaks {}; \ + run 'pilotty stop' so the daemon restarts with the current binary", + response.protocol, PROTOCOL_VERSION + ); + } + + Ok(response) + } + + async fn ensure_protocol( + &mut self, + request_id: &str, + required: u32, + timeout_duration: Duration, + ) -> Result<()> { + loop { + match protocol_action(self.daemon_protocol, required) { + ProtocolAction::Send => return Ok(()), + ProtocolAction::Probe => { + let probe = Request::new( + format!("{request_id}-protocol-probe"), + Command::ListSessions, + ); + let response = self.exchange(probe, timeout_duration).await?; + self.daemon_protocol = Some(response.protocol); + } + ProtocolAction::Reject { observed, required } => { + return Err(ApiError::protocol_upgrade_required(observed, required).into()); + } + } + } + } + + async fn exchange(&mut self, request: Request, timeout_duration: Duration) -> Result { let request_json = serde_json::to_string(&request).context("Failed to serialize request")?; debug!("Sending: {}", request_json); @@ -155,15 +227,6 @@ impl DaemonClient { let response: Response = serde_json::from_str(&response_line).context("Failed to parse response")?; - - if response.protocol < PROTOCOL_VERSION { - eprintln!( - "warning: the running daemon speaks protocol {} but this client speaks {}; \ - run 'pilotty stop' so the daemon restarts with the current binary", - response.protocol, PROTOCOL_VERSION - ); - } - Ok(response) } } @@ -197,7 +260,10 @@ mod tests { let stream = UnixStream::connect(&socket_path) .await .expect("Failed to connect"); - let mut client = DaemonClient { stream }; + let mut client = DaemonClient { + stream, + daemon_protocol: None, + }; // Send request let request = Request { @@ -214,4 +280,28 @@ mod tests { server_handle.abort(); let _ = std::fs::remove_file(&socket_path); } + + #[test] + fn unknown_peer_is_probed_only_for_versioned_commands() { + assert_eq!(protocol_action(None, 0), ProtocolAction::Send); + assert_eq!( + protocol_action(None, PROTOCOL_VERSION), + ProtocolAction::Probe + ); + } + + #[test] + fn observed_peer_protocol_decides_versioned_command_support() { + assert_eq!( + protocol_action(Some(0), PROTOCOL_VERSION), + ProtocolAction::Reject { + observed: 0, + required: PROTOCOL_VERSION, + } + ); + assert_eq!( + protocol_action(Some(PROTOCOL_VERSION), PROTOCOL_VERSION), + ProtocolAction::Send + ); + } } diff --git a/crates/pilotty-cli/src/daemon/server.rs b/crates/pilotty-cli/src/daemon/server.rs index b8309ed..3e4b352 100644 --- a/crates/pilotty-cli/src/daemon/server.rs +++ b/crates/pilotty-cli/src/daemon/server.rs @@ -7,7 +7,9 @@ use std::time::{Duration, Instant}; use anyhow::{Context, Result}; use pilotty_core::error::ApiError; use pilotty_core::input::encode_mouse_click_combined; -use pilotty_core::protocol::{Command, Request, Response, ResponseData, SnapshotFormat}; +use pilotty_core::protocol::{ + supports_protocol, Command, Request, Response, ResponseData, SnapshotFormat, +}; use pilotty_core::snapshot::{CursorState, ScreenState, TerminalSize}; use tokio::io::{AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; @@ -503,14 +505,23 @@ async fn handle_request( sessions: Arc, shutdown: Arc, ) -> Response { - debug!("Handling command: {:?}", request.command); + let request_protocol = request.protocol; + let request_id = request.id; + let command = request.command; + debug!("Handling command: {:?}", command); - match request.command { + if let Some(response) = + protocol_mismatch_response(&request_id, request_protocol, command.minimum_protocol()) + { + return response; + } + + let response = match command { Command::Spawn { command, session_name, cwd, - } => handle_spawn(&request.id, &sessions, command, session_name, cwd).await, + } => handle_spawn(&request_id, &sessions, command, session_name, cwd).await, Command::Snapshot { session, @@ -520,7 +531,7 @@ async fn handle_request( timeout_ms, } => { handle_snapshot( - &request.id, + &request_id, &sessions, session, format, @@ -531,43 +542,57 @@ async fn handle_request( .await } - Command::ListSessions => handle_list_sessions(&request.id, &sessions).await, + Command::ListSessions => handle_list_sessions(&request_id, &sessions).await, - Command::Kill { session } => handle_kill(&request.id, &sessions, session).await, + Command::Kill { session } => handle_kill(&request_id, &sessions, session).await, - Command::Type { text, session } => handle_type(&request.id, &sessions, text, session).await, + Command::Type { text, session } => handle_type(&request_id, &sessions, text, session).await, Command::Key { key, delay_ms, session, - } => handle_key(&request.id, &sessions, key, delay_ms, session).await, + } => handle_key(&request_id, &sessions, key, delay_ms, session).await, Command::Click { row, col, session } => { - handle_click(&request.id, &sessions, row, col, session).await + handle_click(&request_id, &sessions, row, col, session).await } Command::Scroll { direction, amount, session, - } => handle_scroll(&request.id, &sessions, direction, amount, session).await, + } => handle_scroll(&request_id, &sessions, direction, amount, session).await, Command::WaitFor { pattern, timeout_ms, regex, session, - } => handle_wait_for(&request.id, &sessions, pattern, timeout_ms, regex, session).await, + } => handle_wait_for(&request_id, &sessions, pattern, timeout_ms, regex, session).await, Command::Resize { cols, rows, session, - } => handle_resize(&request.id, &sessions, cols, rows, session).await, + } => handle_resize(&request_id, &sessions, cols, rows, session).await, - Command::Shutdown => handle_shutdown(&request.id, sessions, shutdown).await, + Command::Shutdown => handle_shutdown(&request_id, sessions, shutdown).await, + }; + + protocol_mismatch_response(&request_id, request_protocol, response.minimum_protocol()) + .unwrap_or(response) +} + +fn protocol_mismatch_response(request_id: &str, observed: u32, required: u32) -> Option { + if supports_protocol(observed, required) { + return None; } + + Some(Response::error( + request_id, + ApiError::protocol_upgrade_required(observed, required), + )) } /// Handle spawn command. @@ -1410,6 +1435,63 @@ mod tests { let _ = std::fs::remove_file(&socket_path); } + #[tokio::test] + async fn legacy_client_can_use_existing_command_against_current_daemon() { + let short_id = Uuid::new_v4().simple().to_string(); + let socket_path = std::path::PathBuf::from("/tmp") + .join(format!("pilotty-legacy-client-{}.sock", &short_id[..8])); + let pid_path = socket_path.with_extension("pid"); + let server = DaemonServer::bind_to(socket_path.clone(), pid_path.clone()) + .await + .expect("bind server"); + let server_handle = tokio::spawn(async move { + let _ = timeout(Duration::from_secs(2), server.run()).await; + }); + + tokio::time::sleep(Duration::from_millis(50)).await; + let stream = UnixStream::connect(&socket_path) + .await + .expect("connect legacy client"); + let (reader, mut writer) = stream.into_split(); + let mut reader = BufReader::new(reader); + + writer + .write_all(b"{\"id\":\"legacy-1\",\"command\":{\"action\":\"list_sessions\"}}\n") + .await + .expect("write legacy request"); + writer.flush().await.expect("flush legacy request"); + + let mut response_line = String::new(); + timeout(Duration::from_secs(1), reader.read_line(&mut response_line)) + .await + .expect("response timeout") + .expect("read response"); + let response: Response = serde_json::from_str(&response_line).expect("parse response"); + + assert!(response.success); + assert_eq!(response.id, "legacy-1"); + + server_handle.abort(); + let _ = std::fs::remove_file(&socket_path); + let _ = std::fs::remove_file(&pid_path); + } + + #[test] + fn protocol_guard_returns_legacy_compatible_upgrade_error() { + assert!(protocol_mismatch_response("req-1", PROTOCOL_VERSION, PROTOCOL_VERSION).is_none()); + + let response = protocol_mismatch_response("req-2", 0, PROTOCOL_VERSION) + .expect("older peer must receive an upgrade response"); + let error = response.error.expect("upgrade error"); + + assert_eq!(response.id, "req-2"); + assert_eq!(error.code, ErrorCode::InvalidInput); + assert_eq!(error.minimum_protocol(), 0); + assert!(error.suggestion.as_deref().is_some_and(|suggestion| { + suggestion.contains("pilotty stop") && suggestion.contains("retry") + })); + } + #[tokio::test] async fn test_read_line_bounded_handles_utf8_chunks() { let data = "hello 你好\n".as_bytes().to_vec(); diff --git a/crates/pilotty-core/src/error.rs b/crates/pilotty-core/src/error.rs index b84825a..9e5113f 100644 --- a/crates/pilotty-core/src/error.rs +++ b/crates/pilotty-core/src/error.rs @@ -107,6 +107,21 @@ impl ApiError { } } + /// Create a legacy-compatible error for protocol version skew. + pub fn protocol_upgrade_required(observed: u32, required: u32) -> Self { + Self { + code: ErrorCode::InvalidInput, + message: format!( + "Peer speaks protocol {}, but this operation requires protocol {}", + observed, required + ), + suggestion: Some( + "Update pilotty, run 'pilotty stop', and retry so the client and daemon use the same version" + .into(), + ), + } + } + pub fn duplicate_session_name(name: &str) -> Self { Self { code: ErrorCode::InvalidInput, @@ -316,4 +331,17 @@ mod tests { assert_eq!(err.message, "Session 'x' not found"); assert_eq!(err.suggestion, Some("hint".to_string())); } + + #[test] + fn protocol_upgrade_error_is_legacy_compatible_and_actionable() { + let err = ApiError::protocol_upgrade_required(0, 1); + + assert_eq!(err.code, ErrorCode::InvalidInput); + assert_eq!(err.minimum_protocol(), 0); + assert!(err.message.contains("protocol 0")); + assert!(err.message.contains("protocol 1")); + assert!(err.suggestion.as_deref().is_some_and(|suggestion| { + suggestion.contains("pilotty stop") && suggestion.contains("retry") + })); + } } diff --git a/crates/pilotty-core/src/protocol.rs b/crates/pilotty-core/src/protocol.rs index 3c9e04f..3fb2af4 100644 --- a/crates/pilotty-core/src/protocol.rs +++ b/crates/pilotty-core/src/protocol.rs @@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize}; -use crate::error::ApiError; +use crate::error::{ApiError, ErrorCode}; use crate::snapshot::ScreenState; /// Default timeout for snapshot await_change/settle operations (30 seconds). @@ -20,6 +20,29 @@ fn default_snapshot_timeout() -> u64 { /// must treat a lower daemon version as "this command may not exist yet". pub const PROTOCOL_VERSION: u32 = 1; +/// Protocol spoken by binaries that predate explicit versioning. +pub const LEGACY_PROTOCOL_VERSION: u32 = 0; + +/// Whether an observed peer protocol satisfies a wire variant's requirement. +pub fn supports_protocol(observed: u32, required: u32) -> bool { + observed >= required +} + +impl ApiError { + /// Oldest protocol that can decode this error without losing meaning. + /// + /// Keeping the exhaustive error-code mapping with the other wire policy + /// makes a new code declare its compatibility before it can compile. + pub fn minimum_protocol(&self) -> u32 { + match &self.code { + ErrorCode::SessionNotFound + | ErrorCode::CommandFailed + | ErrorCode::InvalidInput + | ErrorCode::InternalError => LEGACY_PROTOCOL_VERSION, + } + } +} + /// A request from CLI to daemon. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct Request { @@ -121,6 +144,28 @@ pub enum Command { Shutdown, } +impl Command { + /// Oldest protocol that can decode and honor this command. + /// + /// Keep this match exhaustive so every new command must declare its wire + /// compatibility before it can compile. + pub fn minimum_protocol(&self) -> u32 { + match self { + Self::Spawn { .. } + | Self::Kill { .. } + | Self::Snapshot { .. } + | Self::Type { .. } + | Self::Key { .. } + | Self::Click { .. } + | Self::Scroll { .. } + | Self::ListSessions + | Self::Resize { .. } + | Self::WaitFor { .. } + | Self::Shutdown => LEGACY_PROTOCOL_VERSION, + } + } +} + /// Snapshot output format. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -176,6 +221,21 @@ impl Response { protocol: PROTOCOL_VERSION, } } + + /// Oldest protocol that can decode this response without losing meaning. + pub fn minimum_protocol(&self) -> u32 { + let data_protocol = self + .data + .as_ref() + .map(ResponseData::minimum_protocol) + .unwrap_or(LEGACY_PROTOCOL_VERSION); + let error_protocol = self + .error + .as_ref() + .map(ApiError::minimum_protocol) + .unwrap_or(LEGACY_PROTOCOL_VERSION); + data_protocol.max(error_protocol) + } } /// Response payload variants. @@ -203,6 +263,23 @@ pub enum ResponseData { Ok { message: String }, } +impl ResponseData { + /// Oldest protocol that can decode this response payload. + /// + /// The exhaustive match prevents a new payload from silently reaching an + /// older client. + pub fn minimum_protocol(&self) -> u32 { + match self { + Self::ScreenState(_) + | Self::Snapshot { .. } + | Self::SessionCreated { .. } + | Self::Sessions { .. } + | Self::WaitForResult { .. } + | Self::Ok { .. } => LEGACY_PROTOCOL_VERSION, + } + } +} + /// Information about an active session. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct SessionInfo { @@ -264,4 +341,47 @@ mod tests { let response: Response = serde_json::from_str(json).unwrap(); assert_eq!(response.protocol, 0); } + + #[test] + fn legacy_client_ignores_unknown_response_protocol_field() { + #[derive(Deserialize)] + struct LegacyResponse { + id: String, + success: bool, + } + + let response = Response::success( + "req-1", + ResponseData::Ok { + message: "done".to_string(), + }, + ); + let json = serde_json::to_string(&response).expect("serialize current response"); + let legacy: LegacyResponse = + serde_json::from_str(&json).expect("legacy client should ignore added fields"); + + assert_eq!(legacy.id, "req-1"); + assert!(legacy.success); + } + + #[test] + fn protocol_support_is_monotonic() { + assert!(supports_protocol(1, 0)); + assert!(supports_protocol(1, 1)); + assert!(!supports_protocol(0, 1)); + } + + #[test] + fn existing_wire_variants_remain_legacy_compatible() { + let command = Command::ListSessions; + let response = Response::success( + "req-1", + ResponseData::Ok { + message: "done".to_string(), + }, + ); + + assert_eq!(command.minimum_protocol(), 0); + assert_eq!(response.minimum_protocol(), 0); + } }