adjusts spinners to be compatible with conduit

Author: dingfeli

Cargo.lock

@@ -19,6 +19,7 @@ serde.workspace = true
 serde_json.workspace = true
 chrono.workspace = true
 crossterm.workspace = true
+tokio.workspace = true
 
 [target.'cfg(unix)'.dependencies]
 nix.workspace = true

crates/chat-cli-ui/Cargo.toml

@@ -21,7 +21,7 @@ pub enum ConduitError {
 pub struct ViewEnd {
     /// Used by the view to send input to the control
     // TODO: later on we will need replace this byte array with an actual event type from ACP
-    pub sender: std::sync::mpsc::Sender<Vec<u8>>,
+    pub sender: tokio::sync::mpsc::Sender<Vec<u8>>,
     /// To receive messages from control about state changes
     pub receiver: std::sync::mpsc::Receiver<Event>,
 }
@@ -61,7 +61,7 @@ pub struct DestinationStderr;
 #[derive(Clone, Debug)]
 pub struct DestinationStructuredOutput;
 
-pub type InputReceiver = std::sync::mpsc::Receiver<Vec<u8>>;
+pub type InputReceiver = tokio::sync::mpsc::Receiver<Vec<u8>>;
 
 /// This compliments the [ViewEnd]. It can be thought of as the "other end" of a pipe.
 /// The control would own this.
@@ -212,7 +212,7 @@ pub fn get_legacy_conduits() -> (
     ControlEnd<DestinationStdout>,
 ) {
     let (state_tx, state_rx) = std::sync::mpsc::channel::<Event>();
-    let (byte_tx, byte_rx) = std::sync::mpsc::channel::<Vec<u8>>();
+    let (byte_tx, byte_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(10);
 
     (
         ViewEnd {

crates/chat-cli-ui/src/conduit.rs

@@ -0,0 +1,54 @@
+use crossterm::{cursor, execute, style, terminal};
+use indicatif::{ProgressBar, ProgressStyle};
+use tokio_util::sync::CancellationToken;
+
+const SPINNER_CHARS: &str = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏";
+
+pub struct Spinners {
+    cancellation_token: CancellationToken,
+}
+
+impl Spinners {
+    pub fn new(message: String) -> Self {
+        let pb = ProgressBar::new_spinner();
+        pb.set_style(
+            ProgressStyle::default_spinner()
+                .tick_chars(SPINNER_CHARS)
+                .template("{spinner:.green} {msg}")
+                .unwrap(),
+        );
+        pb.set_message(message);
+        let token = CancellationToken::new();
+        let token_clone = token.clone();
+        tokio::spawn(async move {
+            loop {
+                tokio::select! {
+                    _ = token_clone.cancelled() => {
+                        break;
+                    },
+                    _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {
+                        pb.tick();
+                    }
+                }
+            }
+
+            Ok::<(), Box<dyn std::error::Error + Send + Sync + 'static>>(())
+        });
+
+        Self {
+            cancellation_token: token,
+        }
+    }
+}
+
+impl Drop for Spinners {
+    fn drop(&mut self) {
+        self.cancellation_token.cancel();
+        let _ = execute!(
+            std::io::stderr(),
+            cursor::MoveToColumn(0),
+            terminal::Clear(terminal::ClearType::CurrentLine),
+            style::Print("\n"),
+        );
+    }
+}

crates/chat-cli/src/cli/chat/custom_spinner.rs

@@ -2,6 +2,7 @@ pub mod cli;
 mod consts;
 pub mod context;
 mod conversation;
+mod custom_spinner;
 mod input_source;
 mod message;
 mod parse;
@@ -36,14 +37,14 @@ pub use conversation::ConversationState;
 use conversation::TokenWarningLevel;
 use crossterm::style::{Attribute, Color, Stylize};
 use crossterm::{cursor, execute, queue, style, terminal};
+use custom_spinner::Spinners;
 use eyre::{Report, Result, bail, eyre};
 use input_source::InputSource;
 use message::{AssistantMessage, AssistantToolUse, ToolUseResult, ToolUseResultBlock};
 use parse::{ParseState, interpret_markdown};
 use parser::{RecvErrorKind, RequestMetadata, SendMessageStream};
 use regex::Regex;
 use rmcp::model::PromptMessage;
-use spinners::{Spinner, Spinners};
 use thiserror::Error;
 use time::OffsetDateTime;
 use token_counter::TokenCounter;
@@ -169,7 +170,6 @@ impl ChatArgs {
             }
         }
 
-        let stdout = std::io::stdout();
         let mut stderr = std::io::stderr();
 
         let args: Vec<String> = std::env::args().collect();
@@ -336,8 +336,6 @@ impl ChatArgs {
 
         ChatSession::new(
             os,
-            stdout,
-            stderr,
             &conversation_id,
             agents,
             input,
@@ -540,7 +538,7 @@ pub struct ChatSession {
     input_source: InputSource,
     /// Width of the terminal, required for [ParseState].
     terminal_width_provider: fn() -> Option<usize>,
-    spinner: Option<Spinner>,
+    spinner: Option<Spinners>,
     /// [ConversationState].
     conversation: ConversationState,
     /// Tool uses requested by the model that are actively being handled.
@@ -572,8 +570,6 @@ impl ChatSession {
     #[allow(clippy::too_many_arguments)]
     pub async fn new(
         os: &mut Os,
-        stdout: std::io::Stdout,
-        mut stderr: std::io::Stderr,
         conversation_id: &str,
         mut agents: Agents,
         mut input: Option<String>,
@@ -594,7 +590,7 @@ impl ChatSession {
             .and_then(|cwd| os.database.get_conversation_by_path(cwd).ok())
             .flatten();
 
-        let (view_end, _byte_receiver, control_end_stderr, control_end_stdout) = get_legacy_conduits();
+        let (view_end, _byte_receiver, mut control_end_stderr, control_end_stdout) = get_legacy_conduits();
 
         tokio::task::spawn_blocking(move || {
             let stderr = std::io::stderr();
@@ -619,7 +615,7 @@ impl ChatSession {
                 if let Some(profile) = cs.current_profile() {
                     if agents.switch(profile).is_err() {
                         execute!(
-                            stderr,
+                            control_end_stderr,
                             style::SetForegroundColor(Color::Red),
                             style::Print("Error"),
                             style::ResetColor,
@@ -1086,10 +1082,6 @@ impl ChatSession {
 
 impl Drop for ChatSession {
     fn drop(&mut self) {
-        if let Some(spinner) = &mut self.spinner {
-            spinner.stop();
-        }
-
         execute!(
             self.stderr,
             cursor::MoveToColumn(0),
@@ -1386,7 +1378,7 @@ impl ChatSession {
 
         if self.interactive {
             execute!(self.stderr, cursor::Hide, style::Print("\n"))?;
-            self.spinner = Some(Spinner::new(Spinners::Dots, "Creating summary...".to_string()));
+            self.spinner = Some(Spinners::new("Creating summary...".to_string()));
         }
 
         let mut response = match self
@@ -1681,10 +1673,10 @@ impl ChatSession {
 
         if self.interactive {
             execute!(self.stderr, cursor::Hide, style::Print("\n"))?;
-            self.spinner = Some(Spinner::new(
-                Spinners::Dots,
-                format!("Generating agent config for '{}'...", agent_name),
-            ));
+            self.spinner = Some(Spinners::new(format!(
+                "Generating agent config for '{}'...",
+                agent_name
+            )));
         }
 
         let mut response = match self
@@ -2132,7 +2124,7 @@ impl ChatSession {
             queue!(self.stderr, cursor::Hide)?;
 
             if self.interactive {
-                self.spinner = Some(Spinner::new(Spinners::Dots, "Thinking...".to_owned()));
+                self.spinner = Some(Spinners::new("Thinking...".to_owned()));
             }
 
             Ok(ChatState::HandleResponseStream(conv_state))
@@ -2537,7 +2529,7 @@ impl ChatSession {
         execute!(self.stderr, cursor::Hide)?;
         execute!(self.stderr, style::Print("\n"), style::SetAttribute(Attribute::Reset))?;
         if self.interactive {
-            self.spinner = Some(Spinner::new(Spinners::Dots, "Thinking...".to_string()));
+            self.spinner = Some(Spinners::new("Thinking...".to_string()));
         }
 
         self.send_chat_telemetry(os, TelemetryResult::Succeeded, None, None, None, false)
@@ -2685,7 +2677,7 @@ impl ChatSession {
                             );
 
                             execute!(self.stderr, cursor::Hide)?;
-                            self.spinner = Some(Spinner::new(Spinners::Dots, "Dividing up the work...".to_string()));
+                            self.spinner = Some(Spinners::new("Dividing up the work...".to_string()));
 
                             // For stream timeouts, we'll tell the model to try and split its response into
                             // smaller chunks.
@@ -2859,7 +2851,7 @@ impl ChatSession {
             if tool_name_being_recvd.is_some() {
                 queue!(self.stderr, cursor::Hide)?;
                 if self.interactive {
-                    self.spinner = Some(Spinner::new(Spinners::Dots, "Thinking...".to_string()));
+                    self.spinner = Some(Spinners::new("Thinking...".to_string()));
                 }
             }
 
@@ -3176,7 +3168,7 @@ impl ChatSession {
         }
 
         if self.interactive {
-            self.spinner = Some(Spinner::new(Spinners::Dots, "Thinking...".to_owned()));
+            self.spinner = Some(Spinners::new("Thinking...".to_owned()));
         }
 
         Ok(ChatState::HandleResponseStream(
@@ -3591,32 +3583,34 @@ async fn get_subscription_status(os: &mut Os) -> Result<ActualSubscriptionStatus
 
 async fn get_subscription_status_with_spinner(
     os: &mut Os,
-    output: &mut impl Write,
+    output: &mut (impl Write + Clone + Send + Sync + 'static),
 ) -> Result<ActualSubscriptionStatus> {
     return with_spinner(output, "Checking subscription status...", || async {
         get_subscription_status(os).await
     })
     .await;
 }
 
-pub async fn with_spinner<T, E, F, Fut>(output: &mut impl std::io::Write, spinner_text: &str, f: F) -> Result<T, E>
+pub async fn with_spinner<T, E, F, Fut, S: std::io::Write + Clone + Send + Sync + 'static>(
+    output: &mut S,
+    spinner_text: &str,
+    f: F,
+) -> Result<T, E>
 where
     F: FnOnce() -> Fut,
     Fut: std::future::Future<Output = Result<T, E>>,
 {
     queue!(output, cursor::Hide,).ok();
-    let spinner = Some(Spinner::new(Spinners::Dots, spinner_text.to_owned()));
+    let spinner = Spinners::new(spinner_text.to_owned());
 
     let result = f().await;
 
-    if let Some(mut s) = spinner {
-        s.stop();
-        let _ = queue!(
-            output,
-            terminal::Clear(terminal::ClearType::CurrentLine),
-            cursor::MoveToColumn(0),
-        );
-    }
+    drop(spinner);
+    let _ = queue!(
+        output,
+        terminal::Clear(terminal::ClearType::CurrentLine),
+        cursor::MoveToColumn(0),
+    );
 
     result
 }
@@ -3727,8 +3721,6 @@ mod tests {
             .expect("Tools failed to load");
         ChatSession::new(
             &mut os,
-            std::io::stdout(),
-            std::io::stderr(),
             "fake_conv_id",
             agents,
             None,
@@ -3857,8 +3849,6 @@ mod tests {
             .expect("Tools failed to load");
         ChatSession::new(
             &mut os,
-            std::io::stdout(),
-            std::io::stderr(),
             "fake_conv_id",
             agents,
             None,
@@ -3964,8 +3954,6 @@ mod tests {
             .expect("Tools failed to load");
         ChatSession::new(
             &mut os,
-            std::io::stdout(),
-            std::io::stderr(),
             "fake_conv_id",
             agents,
             None,
@@ -4042,8 +4030,6 @@ mod tests {
             .expect("Tools failed to load");
         ChatSession::new(
             &mut os,
-            std::io::stdout(),
-            std::io::stderr(),
             "fake_conv_id",
             agents,
             None,
@@ -4100,8 +4086,6 @@ mod tests {
             .expect("Tools failed to load");
         ChatSession::new(
             &mut os,
-            std::io::stdout(),
-            std::io::stderr(),
             "fake_conv_id",
             agents,
             None,
@@ -4208,8 +4192,6 @@ mod tests {
         // Test that PreToolUse hook runs
         ChatSession::new(
             &mut os,
-            std::io::stdout(),
-            std::io::stderr(),
             "fake_conv_id",
             agents,
             None, // No initial input
@@ -4344,8 +4326,6 @@ mod tests {
         // Run chat session - hook should block tool execution
         let result = ChatSession::new(
             &mut os,
-            std::io::stdout(),
-            std::io::stderr(),
             "test_conv_id",
             agents,
             None,