diff --git a/CHANGELOG.md b/CHANGELOG.md
index af76a97..fb3198c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- @JesseGuillory-CM removed several panics and poor error handling from rosbridge client.
+
 ### Changed
 
 ## 0.21.0 - May 19th, 2026
diff --git a/roslibrust_rosbridge/src/client.rs b/roslibrust_rosbridge/src/client.rs
index 57de800..d36d147 100644
--- a/roslibrust_rosbridge/src/client.rs
+++ b/roslibrust_rosbridge/src/client.rs
@@ -346,30 +346,44 @@ impl ClientHandle {
         self.check_for_disconnect()?;
         let (tx, rx) = tokio::sync::oneshot::channel();
         let rand_string: String = uuid::Uuid::new_v4().to_string();
-        let client = self.inner.read().await;
+        let response_timeout;
+        let service_calls;
         {
-            if client
-                .service_calls
-                .insert(rand_string.clone(), tx)
-                .is_some()
-            {
+            let client = self.inner.read().await;
+            // Close the race between the initial check and acquiring the client lock.
+            self.check_for_disconnect()?;
+            response_timeout = client.opts.timeout;
+            service_calls = client.service_calls.clone();
+            if service_calls.insert(rand_string.clone(), tx).is_some() {
                 error!("ID collision encountered in call_service");
             }
-        }
-        {
             let mut comm = client.writer.write().await;
-            timeout(
+            if let Err(e) = timeout(
                 client.opts.timeout,
                 comm.call_service(service, &rand_string, req),
             )
-            .await?;
+            .await
+            {
+                // The request never made it onto the wire, no response is coming
+                service_calls.remove(&rand_string);
+                return Err(e);
+            }
         }
 
+        // Do not retain the outer client lock while waiting for the response.
+        // Reconnection needs its write half to replace the connection.
+
         // Having to do manual timeout logic here because of error types
-        let recv = if let Some(timeout) = client.opts.timeout {
-            tokio::time::timeout(timeout, rx)
-                .await
-                .map_err(|e| Error::Timeout(format!("Service call timed out: {e:?}")))?
+        let recv = if let Some(timeout) = response_timeout {
+            match tokio::time::timeout(timeout, rx).await {
+                Ok(recv) => recv,
+                Err(e) => {
+                    // Stop tracking the call so a late response is discarded
+                    // instead of accumulating in the map
+                    service_calls.remove(&rand_string);
+                    return Err(Error::Timeout(format!("Service call timed out: {e:?}")));
+                }
+            }
         } else {
             rx.await
         };
@@ -377,9 +391,13 @@ impl ClientHandle {
         // Attempt to actually pull data out
         let msg = match recv {
             Ok(msg) => msg,
-            Err(e) =>
-            // TODO remove panic! here, this could result from dropping communication, need to handle disconnect better
-            panic!("The sender end of a service channel was dropped while rx was being awaited, this should not be possible: {}", e),
+            Err(e) => {
+                // The sender was dropped without a response, which happens when the
+                // connection is lost and reconnect() clears the in-flight calls
+                return Err(Error::Unexpected(anyhow!(
+                    "Connection was reset while waiting for a service response: {e}"
+                )));
+            }
         };
 
         // Attempt to convert data to response type
@@ -561,7 +579,7 @@ pub(crate) struct Client {
     services: DashMap<String, ServiceCallback>,
     // Contains any outstanding service calls we're waiting for a response on
     // Map key will be a uniquely generated id for each call
-    service_calls: DashMap<String, tokio::sync::oneshot::Sender<Value>>,
+    service_calls: Arc<DashMap<String, tokio::sync::oneshot::Sender<Value>>>,
     opts: ClientHandleOptions,
 }
 
@@ -575,7 +593,7 @@ impl Client {
             publishers: DashMap::new(),
             services: DashMap::new(),
             subscriptions: DashMap::new(),
-            service_calls: DashMap::new(),
+            service_calls: Arc::new(DashMap::new()),
             opts,
         };
 
@@ -586,17 +604,24 @@ impl Client {
         match msg {
             Message::Text(text) => {
                 debug!("got message: {}", text);
-                // TODO better error handling here serde_json::Error not send
-                let parsed: serde_json::Value = serde_json::from_str(text.as_str()).unwrap();
-                let parsed_object = parsed
-                    .as_object()
-                    .expect("Recieved non-object json response");
-                let op = parsed_object
-                    .get("op")
-                    .expect("Op field not present on returned object.")
-                    .as_str()
-                    .expect("Op field was not of string type.");
-                let op = Ops::from_str(op)?;
+                let parsed: serde_json::Value = match serde_json::from_str(text.as_str()) {
+                    Ok(parsed) => parsed,
+                    Err(e) => {
+                        warn!("Received malformed json message, ignoring: {e}, message: {text}");
+                        return Ok(());
+                    }
+                };
+                let Some(op) = parsed.get("op").and_then(|op| op.as_str()) else {
+                    warn!("Received message without a string `op` field, ignoring: {text}");
+                    return Ok(());
+                };
+                let op = match Ops::from_str(op) {
+                    Ok(op) => op,
+                    Err(e) => {
+                        warn!("Received message with unrecognized op `{op}`, ignoring: {e}");
+                        return Ok(());
+                    }
+                };
                 match op {
                     Ops::Publish => {
                         trace!("handling publish for {:?}", &parsed);
@@ -616,9 +641,11 @@ impl Client {
                 }
             }
             Message::Close(close) => {
-                // TODO how should we respond to this?
-                // How do we represent connection status via our API well?
-                panic!("Close requested from server: {:?}", close);
+                // Returning an error here hands control back to stubborn_spin,
+                // which marks the client disconnected and reconnects.
+                return Err(Error::Unexpected(anyhow!(
+                    "Close requested from server: {close:?}"
+                )));
             }
             Message::Ping(ping) => {
                 debug!("Ping received: {:?}", ping);
@@ -627,7 +654,7 @@ impl Client {
                 debug!("Pong received {:?}", pong);
             }
             _ => {
-                panic!("Non-text response received");
+                warn!("Unexpected non-text response received, ignoring...");
             }
         }
 
@@ -635,51 +662,72 @@ impl Client {
     }
 
     async fn handle_response(&self, data: Value) {
-        // TODO lots of error handling!
-        let id = data.get("id").unwrap().as_str().unwrap();
-        let (_id, call) = self.service_calls.remove(id).unwrap();
-        let res = data.get("values").unwrap();
-        call.send(res.clone()).unwrap();
+        let Some(id) = data.get("id").and_then(|id| id.as_str()) else {
+            warn!("Received service response without a string `id` field, ignoring: {data:?}");
+            return;
+        };
+        let Some((_id, call)) = self.service_calls.remove(id) else {
+            // Can occur legitimately when a response arrives after the caller
+            // stopped waiting (e.g. the caller's timeout expired).
+            warn!("Received service response for unknown or abandoned call id `{id}`, ignoring");
+            return;
+        };
+        let res = data.get("values").cloned().unwrap_or(Value::Null);
+        if call.send(res).is_err() {
+            // The caller gave up on the call (dropped the receiver) before the
+            // response arrived; nothing to deliver it to.
+            debug!("Service response receiver for call id `{id}` was dropped before the response arrived");
+        }
     }
 
     /// Response handler for receiving a service call looks up if we have a service
     /// registered for the incoming topic and if so dispatches to the callback
     async fn handle_service(&self, data: Value) {
-        // Unwrap is okay, field is fully required and strictly type
-        let topic = data.get("service").unwrap().as_str().unwrap();
-        // Unwrap is okay, field is strictly typed to string
-        let id = data.get("id").map(|id| id.as_str().unwrap().to_string());
+        let Some(topic) = data.get("service").and_then(|s| s.as_str()) else {
+            warn!("Received call_service without a string `service` field, ignoring: {data:?}");
+            return;
+        };
+        let id = data
+            .get("id")
+            .and_then(|id| id.as_str())
+            .map(|id| id.to_string());
 
         // Lookup if we have a service for the message
         let callback = self.services.get(topic);
         let callback = match callback {
             Some(callback) => callback,
-            _ => panic!("Received call_service for unadvertised service!"),
+            _ => {
+                warn!("Received call_service for unadvertised service `{topic}`, ignoring");
+                return;
+            }
+        };
+        // TODO likely bugs here. Unclear what we are expected to get for empty service
+        let Some(request) = data.get("args").map(|args| args.to_string()) else {
+            warn!("Received empty service, ignoring...");
+            return;
         };
-        // TODO likely bugs here remove this unwrap. Unclear what we are expected to get for empty service
-        let request = data.get("args").unwrap().to_string();
+
         let mut writer = self.writer.write().await;
 
         // Wrap evaluation of callback in a spawn_blocking to match trait expectations from roslibrust_common
         let callback = callback.value().clone();
         let response = tokio::task::spawn_blocking(move || (callback)(&request))
             .await
-            .expect("Tokio should not cancel or panic in service task");
-        match response {
-            Ok(res) => {
-                // TODO unwrap here is probably bad... Failure to write means disconnected?
-                writer.service_response(topic, id, true, res).await.unwrap();
-            }
+            .unwrap_or_else(|e| Err(format!("Service callback panicked: {e}").into()));
+        // A failed write means the connection dropped; spin_once will notice and
+        // trigger a reconnect, so logging is all we can usefully do here
+        let write_result = match response {
+            Ok(res) => writer.service_response(topic, id, true, res).await,
             Err(e) => {
                 error!("A service callback on topic {:?} failed with {:?} sending response false in service_response", data.get("service"), e);
                 writer
                     .service_response(topic, id, false, serde_json::json!(format!("{e}")))
                     .await
-                    .unwrap();
             }
         };
-
-        // Now we need to send the service_response back
+        if let Err(e) = write_result {
+            error!("Failed to send service_response for `{topic}`: {e}");
+        }
     }
 
     async fn spin_once(&self) -> Result<()> {
@@ -703,25 +751,37 @@ impl Client {
     /// Converts the return message to the subscribed type and calls any callbacks
     /// Panics if publish is received for unexpected topic
     async fn handle_publish(&self, data: Value) {
-        // TODO lots of error handling!
-        let callbacks = self
-            .subscriptions
-            .get(data.get("topic").unwrap().as_str().unwrap());
-        let callbacks = match callbacks {
+        let Some(topic) = data.get("topic").and_then(|t| t.as_str()) else {
+            warn!("Received publish without a string `topic` field, ignoring: {data:?}");
+            return;
+        };
+        let callbacks = match self.subscriptions.get(topic) {
             Some(callbacks) => callbacks,
-            _ => panic!("Received publish message for unsubscribed topic!"), // TODO probably shouldn't be a panic?
+            // Can occur legitimately in the window between rosbridge sending a
+            // message and it processing our unsubscribe
+            _ => {
+                debug!("Received publish for unsubscribed topic `{topic}`, ignoring");
+                return;
+            }
+        };
+        let Some(msg) = data.get("msg") else {
+            warn!("Received publish without a `msg` field on topic `{topic}`, ignoring");
+            return;
+        };
+        let Ok(msg) = serde_json::to_string(msg) else {
+            warn!("Failed to re-serialize `msg` field on topic `{topic}`, ignoring");
+            return;
         };
         for callback in callbacks.handles.values() {
-            callback(
-                // TODO possible bug here if "msg" isn't defined remove this unwrap
-                serde_json::to_string(data.get("msg").unwrap())
-                    .unwrap()
-                    .as_str(),
-            )
+            callback(msg.as_str())
         }
     }
 
     async fn reconnect(&mut self) -> Result<()> {
+        // Responses to calls made on the old connection will never arrive.
+        // This also catches calls that raced with the initial disconnect cleanup.
+        self.service_calls.clear();
+
         // Reconnect stream
         let (writer, reader) = stubborn_connect(&self.opts.url).await;
         self.reader = RwLock::new(reader);
@@ -771,8 +831,16 @@ async fn stubborn_spin(
             Ok(Ok(())) => {}
             Ok(Err(err)) => {
                 is_disconnected.store(true, Ordering::Relaxed);
+                // Dropping the senders wakes in-flight service calls before we
+                // request the exclusive lock needed to reconnect.
+                client.read().await.service_calls.clear();
                 warn!("Spin failed with error: {err}, attempting to reconnect");
-                client.write().await.reconnect().await?;
+                // Never propagate a reconnect failure: this task is the only
+                // thing keeping the connection alive, so keep retrying
+                while let Err(e) = client.write().await.reconnect().await {
+                    warn!("Reconnect attempt failed: {e}, retrying");
+                    tokio::time::sleep(Duration::from_millis(200)).await;
+                }
                 is_disconnected.store(false, Ordering::Relaxed);
             }
             Err(_) => {
@@ -827,3 +895,165 @@ async fn connect(url: &str) -> Result<Socket> {
         Err(e) => Err(Error::IoError(std::io::Error::other(e))),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::SinkExt;
+    use roslibrust_test::ros1::std_srvs::{SetBool, SetBoolRequest};
+    use tokio::net::TcpListener;
+    use tokio_tungstenite::WebSocketStream;
+
+    type ServerSocket = WebSocketStream<tokio::net::TcpStream>;
+
+    // Create a dummy no-op server and connect a client to it.
+    async fn test_connection(
+        client_timeout: Option<Duration>,
+    ) -> (TcpListener, ClientHandle, ServerSocket) {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+        let mut opts = ClientHandleOptions::new(format!("ws://{address}"));
+        if let Some(client_timeout) = client_timeout {
+            opts = opts.timeout(client_timeout);
+        }
+
+        let connect_client = ClientHandle::new_with_options(opts);
+        let accept_client = async {
+            let (stream, _) = listener.accept().await.unwrap();
+            tokio_tungstenite::accept_async(stream).await.unwrap()
+        };
+        let (client, websocket) = tokio::join!(connect_client, accept_client);
+
+        (listener, client.unwrap(), websocket)
+    }
+
+    async fn next_json(websocket: &mut ServerSocket) -> Value {
+        let message = tokio::time::timeout(Duration::from_secs(2), websocket.next())
+            .await
+            .expect("timed out waiting for client message")
+            .expect("client closed before sending a message")
+            .expect("failed to read client message");
+        let Message::Text(text) = message else {
+            panic!("expected text message, got {message:?}");
+        };
+        serde_json::from_str(&text).unwrap()
+    }
+
+    #[tokio::test]
+    async fn disconnect_wakes_unbounded_service_call() {
+        let (listener, client, mut websocket) = test_connection(None).await;
+        let server = tokio::spawn(async move {
+            let request = next_json(&mut websocket).await;
+            assert_eq!(request["op"], "call_service");
+            websocket.send(Message::Close(None)).await.unwrap();
+            drop(websocket);
+
+            // Keep the listener alive and complete the reconnect handshake. If
+            // the service call retains the client read lock, this never occurs.
+            let (stream, _) = listener.accept().await.unwrap();
+            tokio_tungstenite::accept_async(stream).await.unwrap()
+        });
+
+        // No ClientHandle timeout is configured: only disconnect detection can
+        // resolve the service call.
+        let result = tokio::time::timeout(
+            Duration::from_secs(2),
+            client.call_service::<SetBool>("/test", SetBoolRequest { data: true }),
+        )
+        .await
+        .expect("service call remained blocked after the connection closed");
+
+        assert!(matches!(result, Err(Error::Unexpected(_))));
+        tokio::time::timeout(Duration::from_secs(2), server)
+            .await
+            .expect("client did not reconnect after the service call was released")
+            .unwrap();
+    }
+
+    #[tokio::test]
+    async fn malformed_and_unsupported_messages_are_ignored() {
+        let (_listener, client, _websocket) = test_connection(None).await;
+        let messages = [
+            Message::Text("not json".to_string()),
+            Message::Text("null".to_string()),
+            Message::Text(r#"{"op":42}"#.to_string()),
+            Message::Text(r#"{"op":"unknown"}"#.to_string()),
+            Message::Text(r#"{"op":"publish","msg":{}}"#.to_string()),
+            Message::Text(r#"{"op":"publish","topic":"/test"}"#.to_string()),
+            Message::Text(r#"{"op":"service_response"}"#.to_string()),
+            Message::Text(r#"{"op":"call_service"}"#.to_string()),
+            Message::Binary(vec![1, 2, 3]),
+        ];
+
+        let client = client.inner.read().await;
+        for message in messages {
+            assert!(client.handle_message(message).await.is_ok());
+        }
+        assert!(client.handle_message(Message::Close(None)).await.is_err());
+    }
+
+    #[tokio::test]
+    async fn service_call_timeout_removes_tracking_and_late_response_is_ignored() {
+        let (_listener, client, mut websocket) =
+            test_connection(Some(Duration::from_millis(100))).await;
+        let call_client = client.clone();
+        let call = tokio::spawn(async move {
+            call_client
+                .call_service::<SetBool>("/test", SetBoolRequest { data: true })
+                .await
+        });
+
+        let request = next_json(&mut websocket).await;
+        let id = request["id"].as_str().unwrap().to_string();
+        let result = call.await.unwrap();
+
+        assert!(matches!(result, Err(Error::Timeout(_))));
+        assert!(client.inner.read().await.service_calls.is_empty());
+
+        client
+            .inner
+            .read()
+            .await
+            .handle_response(serde_json::json!({
+                "op": "service_response",
+                "id": id,
+                "values": {"success": true, "message": "too late"}
+            }))
+            .await;
+        assert!(client.inner.read().await.service_calls.is_empty());
+    }
+
+    #[tokio::test]
+    async fn panicking_service_callback_sends_failed_response() {
+        let (_listener, client, mut websocket) = test_connection(None).await;
+        let callback: ServiceCallback = Arc::new(|_| panic!("test callback panic"));
+        client
+            .inner
+            .read()
+            .await
+            .services
+            .insert("/test".to_string(), callback);
+
+        client
+            .inner
+            .read()
+            .await
+            .handle_service(serde_json::json!({
+                "op": "call_service",
+                "service": "/test",
+                "id": "request-id",
+                "args": {"data": true}
+            }))
+            .await;
+
+        let response = next_json(&mut websocket).await;
+        assert_eq!(response["op"], "service_response");
+        assert_eq!(response["service"], "/test");
+        assert_eq!(response["id"], "request-id");
+        assert_eq!(response["result"], false);
+        assert!(response["values"]
+            .as_str()
+            .unwrap()
+            .contains("Service callback panicked"));
+    }
+}