feat: add structured IPC error types with miette diagnostics

src/error.rs

@@ -121,6 +121,64 @@ pub enum FileError {
     NoPath,
 }
 
+/// Errors related to IPC communication with the supervisor.
+#[derive(Debug, Error, Diagnostic)]
+pub enum IpcError {
+    #[error("failed to connect to supervisor after {attempts} attempts")]
+    #[diagnostic(code(pitchfork::ipc::connection_failed))]
+    ConnectionFailed {
+        attempts: u32,
+        #[help]
+        details: Option<String>,
+    },
+
+    #[error("IPC request timed out after {seconds}s")]
+    #[diagnostic(
+        code(pitchfork::ipc::timeout),
+        help(
+            "the supervisor may be unresponsive or overloaded.\nCheck supervisor status: pitchfork supervisor status\nView logs: pitchfork logs"
+        )
+    )]
+    Timeout { seconds: u64 },
+
+    #[error("IPC connection closed unexpectedly")]
+    #[diagnostic(
+        code(pitchfork::ipc::connection_closed),
+        help(
+            "the supervisor may have crashed or been stopped.\nRestart with: pitchfork supervisor start"
+        )
+    )]
+    ConnectionClosed,
+
+    #[error("failed to read IPC response")]
+    #[diagnostic(code(pitchfork::ipc::read_failed))]
+    ReadFailed {
+        #[help]
+        details: Option<String>,
+    },
+
+    #[error("failed to send IPC request")]
+    #[diagnostic(code(pitchfork::ipc::send_failed))]
+    SendFailed {
+        #[help]
+        details: Option<String>,
+    },
+
+    #[error("unexpected response from supervisor: expected {expected}, got {actual}")]
+    #[diagnostic(
+        code(pitchfork::ipc::unexpected_response),
+        help("this may indicate a version mismatch between the CLI and supervisor")
+    )]
+    UnexpectedResponse { expected: String, actual: String },
+
+    #[error("IPC message contains invalid data")]
+    #[diagnostic(code(pitchfork::ipc::invalid_message))]
+    InvalidMessage {
+        #[help]
+        details: Option<String>,
+    },
+}
+
 /// Find the most similar daemon name for suggestions.
 pub fn find_similar_daemon<'a>(
     name: &str,
@@ -217,4 +275,26 @@ mod tests {
         let err = FileError::NoPath;
         assert!(err.to_string().contains("no file path"));
     }
+
+    #[test]
+    fn test_ipc_error_display() {
+        let err = IpcError::ConnectionFailed {
+            attempts: 5,
+            details: Some("connection refused".to_string()),
+        };
+        assert!(err.to_string().contains("failed to connect"));
+        assert!(err.to_string().contains("5 attempts"));
+
+        let err = IpcError::Timeout { seconds: 30 };
+        assert!(err.to_string().contains("timed out"));
+        assert!(err.to_string().contains("30s"));
+
+        let err = IpcError::UnexpectedResponse {
+            expected: "Ok".to_string(),
+            actual: "Error".to_string(),
+        };
+        assert!(err.to_string().contains("unexpected response"));
+        assert!(err.to_string().contains("Ok"));
+        assert!(err.to_string().contains("Error"));
+    }
 }

src/ipc/client.rs

@@ -1,10 +1,11 @@
 use crate::daemon::{Daemon, RunOptions};
+use crate::error::IpcError;
 use crate::ipc::{IpcRequest, IpcResponse, deserialize, fs_name, serialize};
 use crate::{Result, supervisor};
 use exponential_backoff::Backoff;
 use interprocess::local_socket::tokio::{RecvHalf, SendHalf};
 use interprocess::local_socket::traits::tokio::Stream;
-use miette::{Context, IntoDiagnostic, bail, ensure};
+use miette::Context;
 use std::path::PathBuf;
 use std::time::Duration;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
@@ -31,7 +32,13 @@ impl IpcClient {
         let client = Self::connect_(&id, "main").await?;
         trace!("Connected to IPC socket");
         let rsp = client.request(IpcRequest::Connect).await?;
-        ensure!(rsp.is_ok(), "Failed to connect to IPC main");
+        if !rsp.is_ok() {
+            return Err(IpcError::UnexpectedResponse {
+                expected: "Ok".to_string(),
+                actual: format!("{:?}", rsp),
+            }
+            .into());
+        }
         debug!("Connected to IPC main");
         Ok(client)
     }
@@ -58,28 +65,41 @@ impl IpcClient {
                         tokio::time::sleep(duration).await;
                         continue;
                     } else {
-                        bail!("Failed to connect to IPC socket: {:?}", err);
+                        return Err(IpcError::ConnectionFailed {
+                            attempts: CONNECT_ATTEMPTS,
+                            details: Some(format!(
+                                "{err}\nensure the supervisor is running with: pitchfork supervisor start"
+                            )),
+                        }
+                        .into());
                     }
                 }
             }
         }
-        bail!(
-            "failed to connect to IPC socket after {} attempts",
-            CONNECT_ATTEMPTS
-        )
+        Err(IpcError::ConnectionFailed {
+            attempts: CONNECT_ATTEMPTS,
+            details: Some(
+                "ensure the supervisor is running with: pitchfork supervisor start".to_string(),
+            ),
+        }
+        .into())
     }
 
     pub async fn send(&self, msg: IpcRequest) -> Result<()> {
         let mut msg = serialize(&msg)?;
         if msg.contains(&0) {
-            bail!("IPC message contains null byte");
+            return Err(IpcError::InvalidMessage {
+                details: Some("message contains null byte".to_string()),
+            }
+            .into());
         }
         msg.push(0);
         let mut send = self.send.lock().await;
         send.write_all(&msg)
             .await
-            .into_diagnostic()
-            .wrap_err("failed to send IPC message")?;
+            .map_err(|e| IpcError::SendFailed {
+                details: Some(e.to_string()),
+            })?;
         Ok(())
     }
 
@@ -88,11 +108,21 @@ impl IpcClient {
         let mut bytes = Vec::new();
         match tokio::time::timeout(timeout, recv.read_until(0, &mut bytes)).await {
             Ok(Ok(_)) => {}
-            Ok(Err(err)) => bail!("failed to read IPC message: {err}"),
-            Err(_) => bail!("IPC read timed out after {timeout:?}"),
+            Ok(Err(err)) => {
+                return Err(IpcError::ReadFailed {
+                    details: Some(err.to_string()),
+                }
+                .into());
+            }
+            Err(_) => {
+                return Err(IpcError::Timeout {
+                    seconds: timeout.as_secs(),
+                }
+                .into());
+            }
         }
         if bytes.is_empty() {
-            bail!("IPC connection closed unexpectedly (supervisor may have stopped)");
+            return Err(IpcError::ConnectionClosed.into());
         }
         deserialize(&bytes).wrap_err("failed to deserialize IPC response")
     }
@@ -101,6 +131,13 @@ impl IpcClient {
         self.request_with_timeout(msg, REQUEST_TIMEOUT).await
     }
 
+    fn unexpected_response(expected: &str, actual: &IpcResponse) -> IpcError {
+        IpcError::UnexpectedResponse {
+            expected: expected.to_string(),
+            actual: format!("{:?}", actual),
+        }
+    }
+
     async fn request_with_timeout(
         &self,
         msg: IpcRequest,
@@ -121,7 +158,7 @@ impl IpcClient {
                 info!("daemon {} already enabled", id);
                 Ok(false)
             }
-            rsp => bail!("unexpected IPC response: {rsp:?}"),
+            rsp => Err(Self::unexpected_response("Yes or No", &rsp).into()),
         }
     }
 
@@ -136,7 +173,7 @@ impl IpcClient {
                 info!("daemon {} already disabled", id);
                 Ok(false)
             }
-            rsp => bail!("unexpected IPC response: {rsp:?}"),
+            rsp => Err(Self::unexpected_response("Yes or No", &rsp).into()),
         }
     }
 
@@ -185,7 +222,9 @@ impl IpcClient {
                     error!("Failed to print logs: {}", e);
                 }
             }
-            rsp => bail!("unexpected IPC response: {rsp:?}"),
+            rsp => {
+                return Err(Self::unexpected_response("DaemonStart or DaemonReady", &rsp).into());
+            }
         }
         Ok((started_daemons, exit_code))
     }
@@ -194,7 +233,7 @@ impl IpcClient {
         let rsp = self.request(IpcRequest::GetActiveDaemons).await?;
         match rsp {
             IpcResponse::ActiveDaemons(daemons) => Ok(daemons),
-            rsp => bail!("unexpected IPC response: {rsp:?}"),
+            rsp => Err(Self::unexpected_response("ActiveDaemons", &rsp).into()),
         }
     }
 
@@ -209,7 +248,7 @@ impl IpcClient {
             IpcResponse::Ok => {
                 trace!("updated shell dir for pid {shell_pid} to {}", dir.display());
             }
-            rsp => bail!("unexpected IPC response: {rsp:?}"),
+            rsp => return Err(Self::unexpected_response("Ok", &rsp).into()),
         }
         Ok(())
     }
@@ -220,7 +259,7 @@ impl IpcClient {
             IpcResponse::Ok => {
                 trace!("cleaned");
             }
-            rsp => bail!("unexpected IPC response: {rsp:?}"),
+            rsp => return Err(Self::unexpected_response("Ok", &rsp).into()),
         }
         Ok(())
     }
@@ -229,15 +268,15 @@ impl IpcClient {
         let rsp = self.request(IpcRequest::GetDisabledDaemons).await?;
         match rsp {
             IpcResponse::DisabledDaemons(daemons) => Ok(daemons),
-            rsp => bail!("unexpected IPC response: {rsp:?}"),
+            rsp => Err(Self::unexpected_response("DisabledDaemons", &rsp).into()),
         }
     }
 
     pub async fn get_notifications(&self) -> Result<Vec<(log::LevelFilter, String)>> {
         let rsp = self.request(IpcRequest::GetNotifications).await?;
         match rsp {
             IpcResponse::Notifications(notifications) => Ok(notifications),
-            rsp => bail!("unexpected IPC response: {rsp:?}"),
+            rsp => Err(Self::unexpected_response("Notifications", &rsp).into()),
         }
     }
 
@@ -252,7 +291,7 @@ impl IpcClient {
                 warn!("daemon {} is not running", id);
                 Ok(())
             }
-            rsp => bail!("unexpected IPC response: {rsp:?}"),
+            rsp => Err(Self::unexpected_response("Ok or DaemonAlreadyStopped", &rsp).into()),
         }
     }
 }

src/ipc/mod.rs

@@ -2,7 +2,7 @@ use crate::Result;
 use crate::daemon::{Daemon, RunOptions};
 use crate::env;
 use interprocess::local_socket::{GenericFilePath, Name, ToFsName};
-use miette::IntoDiagnostic;
+use miette::{Context, IntoDiagnostic};
 use std::path::PathBuf;
 
 pub(crate) mod client;
@@ -60,22 +60,29 @@ fn fs_name(name: &str) -> Result<Name<'_>> {
 }
 
 fn serialize<T: serde::Serialize>(msg: &T) -> Result<Vec<u8>> {
-    let msg = if *env::IPC_JSON {
-        serde_json::to_vec(msg).into_diagnostic()?
+    if *env::IPC_JSON {
+        serde_json::to_vec(msg)
+            .into_diagnostic()
+            .wrap_err("failed to serialize IPC message as JSON")
     } else {
-        rmp_serde::to_vec(msg).into_diagnostic()?
-    };
-    Ok(msg)
+        rmp_serde::to_vec(msg)
+            .into_diagnostic()
+            .wrap_err("failed to serialize IPC message as MessagePack")
+    }
 }
 
 fn deserialize<T: serde::de::DeserializeOwned>(bytes: &[u8]) -> Result<T> {
     let mut bytes = bytes.to_vec();
     bytes.pop();
-    trace!("msg: {:?}", std::str::from_utf8(&bytes).unwrap_or_default());
-    let msg = if *env::IPC_JSON {
-        serde_json::from_slice(&bytes).into_diagnostic()?
+    let preview = std::str::from_utf8(&bytes).unwrap_or("<binary>");
+    trace!("msg: {:?}", preview);
+    if *env::IPC_JSON {
+        serde_json::from_slice(&bytes)
+            .into_diagnostic()
+            .wrap_err("failed to deserialize IPC JSON response")
     } else {
-        rmp_serde::from_slice(&bytes).into_diagnostic()?
-    };
-    Ok(msg)
+        rmp_serde::from_slice(&bytes)
+            .into_diagnostic()
+            .wrap_err("failed to deserialize IPC MessagePack response")
+    }
 }