Add new --follow mode for tower apps logs with resilient streaming and tests (#171)

* feat (cli): add follow mode for apps logs

* fix(cli): improve follow retry handling

* refactor: reset follow backoff and improve log errors

* reset monitor failures and cancel follow watchers

* harden apps logs follow and add regression tests for

- deduplicate log lines across reconnects / might be an issue
- add unit test for deduplication and out of order logs

* fix linter issues

* wait for run start before log streaming

Author: burakdede

crates/tower-cmd/src/api.rs

@@ -367,6 +367,17 @@ pub enum LogStreamError {
     Unknown,
 }
 
+impl std::fmt::Display for LogStreamError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            LogStreamError::Reqwest(err) => write!(f, "{err}"),
+            LogStreamError::Unknown => write!(f, "unknown log stream error"),
+        }
+    }
+}
+
+impl std::error::Error for LogStreamError {}
+
 impl From<reqwest_eventsource::CannotCloneRequestError> for LogStreamError {
     fn from(err: reqwest_eventsource::CannotCloneRequestError) -> Self {
         debug!("Failed to clone request {:?}", err);
@@ -399,10 +410,30 @@ async fn drain_run_logs_stream(mut source: EventSource, tx: mpsc::Sender<LogStre
                     }
                     "warning" => {
                         let event_warning = serde_json::from_str(&message.data);
-                        if let Ok(event) = event_warning {
-                            tx.send(LogStreamEvent::EventWarning(event)).await.ok();
-                        } else {
-                            debug!("Failed to parse warning message: {:?}", message.data);
+                        match event_warning {
+                            Ok(event) => {
+                                tx.send(LogStreamEvent::EventWarning(event)).await.ok();
+                            }
+                            Err(err) => {
+                                let warning_data = serde_json::from_str(&message.data);
+                                match warning_data {
+                                    Ok(data) => {
+                                        let event = tower_api::models::EventWarning {
+                                            data,
+                                            event: tower_api::models::event_warning::Event::Warning,
+                                            id: None,
+                                            retry: None,
+                                        };
+                                        tx.send(LogStreamEvent::EventWarning(event)).await.ok();
+                                    }
+                                    Err(_) => {
+                                        debug!(
+                                            "Failed to parse warning message: {:?}. Error: {}",
+                                            message.data, err
+                                        );
+                                    }
+                                }
+                            }
                         }
                     }
                     _ => {