🚀 Unleash real-time streaming for agent task execution in Studio

Forge a new frontier in AI interaction with buttery-smooth streaming callbacks
that bring LLM output to life in the TUI as it happens.

The new execute_task_streaming method on IrisAgent opens the floodgates for
real-time token delivery, transforming static waits into dynamic conversations.
This is how AI should feel—responsive, alive, and instantaneous.

Author: hyperb1iss

src/agents/iris.rs

@@ -804,6 +804,125 @@ Guidelines:
         }
     }
 
+    /// Execute a task with streaming, calling the callback with each text chunk
+    ///
+    /// This enables real-time display of LLM output in the TUI.
+    /// The callback receives `(chunk, aggregated_text)` for each delta.
+    ///
+    /// Returns the final structured response after streaming completes.
+    pub async fn execute_task_streaming<F>(
+        &mut self,
+        capability: &str,
+        user_prompt: &str,
+        mut on_chunk: F,
+    ) -> Result<StructuredResponse>
+    where
+        F: FnMut(&str, &str) + Send,
+    {
+        use crate::agents::status::IrisPhase;
+        use crate::messages::get_capability_message;
+        use futures::StreamExt;
+        use rig::agent::MultiTurnStreamItem;
+        use rig::streaming::{StreamedAssistantContent, StreamingPrompt};
+
+        // Show initializing status
+        let waiting_msg = get_capability_message(capability);
+        crate::iris_status_dynamic!(IrisPhase::Initializing, waiting_msg.text, 1, 4);
+
+        // Load the capability config
+        let (mut system_prompt, output_type) = self.load_capability_config(capability)?;
+
+        // Inject style instructions
+        self.inject_style_instructions(&mut system_prompt, capability);
+
+        // Set current capability
+        self.current_capability = Some(capability.to_string());
+
+        // Update status
+        crate::iris_status_dynamic!(
+            IrisPhase::Analysis,
+            "🔍 Iris is analyzing your changes...",
+            2,
+            4
+        );
+
+        // Build the agent
+        let agent = std::sync::Arc::new(self.build_agent()?);
+
+        // Build the full prompt (simplified for streaming - no JSON schema enforcement)
+        let full_prompt = format!(
+            "{}\n\n{}\n\n\
+            After using the available tools, respond with your analysis in markdown format.\n\
+            Keep it clear, well-structured, and informative.",
+            system_prompt, user_prompt
+        );
+
+        // Update status
+        let gen_msg = get_capability_message(capability);
+        crate::iris_status_dynamic!(IrisPhase::Generation, gen_msg.text, 3, 4);
+
+        // Use streaming prompt
+        let mut stream = agent.stream_prompt(&full_prompt).multi_turn(50).await;
+
+        let mut aggregated_text = String::new();
+
+        // Consume the stream
+        while let Some(item) = stream.next().await {
+            match item {
+                Ok(MultiTurnStreamItem::StreamAssistantItem(StreamedAssistantContent::Text(
+                    text,
+                ))) => {
+                    aggregated_text.push_str(&text.text);
+                    on_chunk(&text.text, &aggregated_text);
+                }
+                Ok(MultiTurnStreamItem::FinalResponse(_)) => {
+                    // Stream complete
+                    break;
+                }
+                Err(e) => {
+                    return Err(anyhow::anyhow!("Streaming error: {}", e));
+                }
+                _ => {
+                    // Tool calls, reasoning, etc. - continue
+                }
+            }
+        }
+
+        // Update status
+        crate::iris_status_dynamic!(
+            IrisPhase::Synthesis,
+            "✨ Iris is synthesizing results...",
+            4,
+            4
+        );
+
+        // Convert the aggregated text to structured response based on output type
+        let response = match output_type.as_str() {
+            "MarkdownReview" => StructuredResponse::MarkdownReview(crate::types::MarkdownReview {
+                content: aggregated_text,
+            }),
+            "MarkdownPullRequest" => {
+                StructuredResponse::PullRequest(crate::types::MarkdownPullRequest {
+                    content: aggregated_text,
+                })
+            }
+            "MarkdownChangelog" => StructuredResponse::Changelog(crate::types::MarkdownChangelog {
+                content: aggregated_text,
+            }),
+            "MarkdownReleaseNotes" => {
+                StructuredResponse::ReleaseNotes(crate::types::MarkdownReleaseNotes {
+                    content: aggregated_text,
+                })
+            }
+            "SemanticBlame" => StructuredResponse::SemanticBlame(aggregated_text),
+            _ => StructuredResponse::PlainText(aggregated_text),
+        };
+
+        crate::iris_status_completed!();
+
+        Ok(response)
+    }
+
     /// Load capability configuration from embedded TOML, returning both prompt and output type
     fn load_capability_config(&self, capability: &str) -> Result<(String, String)> {
         let _ = self; // Keep &self for method syntax consistency

src/agents/setup.rs

@@ -372,6 +372,34 @@ impl IrisAgentService {
         agent.execute_task("chat", task_prompt).await
     }
 
+    /// Execute an agent task with streaming
+    ///
+    /// This method streams LLM output in real-time, calling the callback with each
+    /// text chunk as it arrives. Ideal for TUI display of generation progress.
+    ///
+    /// # Arguments
+    /// * `capability` - The capability to invoke (e.g., "review", "pr", "changelog")
+    /// * `context` - Structured context describing what to analyze
+    /// * `on_chunk` - Callback receiving `(chunk, aggregated_text)` for each delta
+    ///
+    /// # Returns
+    /// The final structured response after streaming completes
+    pub async fn execute_task_streaming<F>(
+        &self,
+        capability: &str,
+        context: TaskContext,
+        on_chunk: F,
+    ) -> Result<StructuredResponse>
+    where
+        F: FnMut(&str, &str) + Send,
+    {
+        let mut agent = self.create_agent()?;
+        let task_prompt = Self::build_task_prompt(capability, &context);
+        agent
+            .execute_task_streaming(capability, &task_prompt, on_chunk)
+            .await
+    }
+
     /// Get the configuration
     pub fn config(&self) -> &Config {
         &self.config

src/studio/app.rs

@@ -940,7 +940,7 @@ Simply call the appropriate tool with the new content. Do NOT echo back the full
         }
     }
 
-    /// Spawn a task for code review generation
+    /// Spawn a task for code review generation with streaming
     fn spawn_review_generation(&self, from_ref: String, to_ref: String) {
         use crate::agents::{StructuredResponse, TaskContext};
 
@@ -953,6 +953,7 @@ Simply call the appropriate tool with the new content. Do NOT echo back the full
         };
 
         let tx = self.iris_result_tx.clone();
+        let streaming_tx = tx.clone();
 
         tokio::spawn(async move {
             // Use review context with specified refs
@@ -964,9 +965,29 @@ Simply call the appropriate tool with the new content. Do NOT echo back the full
                 }
             };
 
-            // Execute the review capability
-            match agent.execute_task("review", context).await {
+            // Execute with streaming - send chunks as they arrive
+            let on_chunk = {
+                let tx = streaming_tx.clone();
+                move |chunk: &str, aggregated: &str| {
+                    let _ = tx.send(IrisTaskResult::StreamingChunk {
+                        task_type: TaskType::Review,
+                        chunk: chunk.to_string(),
+                        aggregated: aggregated.to_string(),
+                    });
+                }
+            };
+
+            match agent
+                .execute_task_streaming("review", context, on_chunk)
+                .await
+            {
                 Ok(response) => {
+                    // Send streaming complete
+                    let _ = tx.send(IrisTaskResult::StreamingComplete {
+                        task_type: TaskType::Review,
+                    });
+
+                    // Also send final content
                     let review_text = match response {
                         StructuredResponse::MarkdownReview(review) => review.content,
                         StructuredResponse::PlainText(text) => text,
@@ -981,7 +1002,7 @@ Simply call the appropriate tool with the new content. Do NOT echo back the full
         });
     }
 
-    /// Spawn a task for PR description generation
+    /// Spawn a task for PR description generation with streaming
     fn spawn_pr_generation(&self, base_branch: String, _to_ref: String) {
         use crate::agents::{StructuredResponse, TaskContext};
 
@@ -994,14 +1015,30 @@ Simply call the appropriate tool with the new content. Do NOT echo back the full
         };
 
         let tx = self.iris_result_tx.clone();
+        let streaming_tx = tx.clone();
 
         tokio::spawn(async move {
             // Build context for PR (comparing current branch to base)
             let context = TaskContext::for_pr(Some(base_branch), None);
 
-            // Execute the PR capability
-            match agent.execute_task("pr", context).await {
+            // Execute with streaming
+            let on_chunk = {
+                let tx = streaming_tx.clone();
+                move |chunk: &str, aggregated: &str| {
+                    let _ = tx.send(IrisTaskResult::StreamingChunk {
+                        task_type: TaskType::PR,
+                        chunk: chunk.to_string(),
+                        aggregated: aggregated.to_string(),
+                    });
+                }
+            };
+
+            match agent.execute_task_streaming("pr", context, on_chunk).await {
                 Ok(response) => {
+                    let _ = tx.send(IrisTaskResult::StreamingComplete {
+                        task_type: TaskType::PR,
+                    });
+
                     let pr_text = match response {
                         StructuredResponse::PullRequest(pr) => pr.content,
                         StructuredResponse::PlainText(text) => text,
@@ -1016,7 +1053,7 @@ Simply call the appropriate tool with the new content. Do NOT echo back the full
         });
     }
 
-    /// Spawn a task for changelog generation
+    /// Spawn a task for changelog generation with streaming
     fn spawn_changelog_generation(&self, from_ref: String, to_ref: String) {
         use crate::agents::{StructuredResponse, TaskContext};
 
@@ -1029,14 +1066,33 @@ Simply call the appropriate tool with the new content. Do NOT echo back the full
         };
 
         let tx = self.iris_result_tx.clone();
+        let streaming_tx = tx.clone();
 
         tokio::spawn(async move {
             // Build context for changelog (comparing two refs)
             let context = TaskContext::for_changelog(from_ref, Some(to_ref));
 
-            // Execute the changelog capability
-            match agent.execute_task("changelog", context).await {
+            // Execute with streaming
+            let on_chunk = {
+                let tx = streaming_tx.clone();
+                move |chunk: &str, aggregated: &str| {
+                    let _ = tx.send(IrisTaskResult::StreamingChunk {
+                        task_type: TaskType::Changelog,
+                        chunk: chunk.to_string(),
+                        aggregated: aggregated.to_string(),
+                    });
+                }
+            };
+
+            match agent
+                .execute_task_streaming("changelog", context, on_chunk)
+                .await
+            {
                 Ok(response) => {
+                    let _ = tx.send(IrisTaskResult::StreamingComplete {
+                        task_type: TaskType::Changelog,
+                    });
+
                     let changelog_text = match response {
                         StructuredResponse::Changelog(cl) => cl.content,
                         StructuredResponse::PlainText(text) => text,
@@ -1051,7 +1107,7 @@ Simply call the appropriate tool with the new content. Do NOT echo back the full
         });
     }
 
-    /// Spawn a task for release notes generation
+    /// Spawn a task for release notes generation with streaming
     fn spawn_release_notes_generation(&self, from_ref: String, to_ref: String) {
         use crate::agents::{StructuredResponse, TaskContext};
 
@@ -1064,14 +1120,33 @@ Simply call the appropriate tool with the new content. Do NOT echo back the full
         };
 
         let tx = self.iris_result_tx.clone();
+        let streaming_tx = tx.clone();
 
         tokio::spawn(async move {
             // Build context for release notes (comparing two refs)
             let context = TaskContext::for_changelog(from_ref, Some(to_ref));
 
-            // Execute the release_notes capability
-            match agent.execute_task("release_notes", context).await {
+            // Execute with streaming
+            let on_chunk = {
+                let tx = streaming_tx.clone();
+                move |chunk: &str, aggregated: &str| {
+                    let _ = tx.send(IrisTaskResult::StreamingChunk {
+                        task_type: TaskType::ReleaseNotes,
+                        chunk: chunk.to_string(),
+                        aggregated: aggregated.to_string(),
+                    });
+                }
+            };
+
+            match agent
+                .execute_task_streaming("release_notes", context, on_chunk)
+                .await
+            {
                 Ok(response) => {
+                    let _ = tx.send(IrisTaskResult::StreamingComplete {
+                        task_type: TaskType::ReleaseNotes,
+                    });
+
                     let release_notes_text = match response {
                         StructuredResponse::ReleaseNotes(rn) => rn.content,
                         StructuredResponse::PlainText(text) => text,