feat(protocol): enforce command capability requirements

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<u32>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum ProtocolAction {
+    Send,
+    Probe,
+    Reject { observed: u32, required: u32 },
+}
+
+fn protocol_action(observed: Option<u32>, 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<Response> {
+        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<Response> {
         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
+        );
+    }
 }

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<SessionManager>,
     shutdown: Arc<Notify>,
 ) -> 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<Response> {
+    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();

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")
+        }));
+    }
 }