feat(llm): add real-time token streaming through provider chain

src/agent/dispatcher.rs

@@ -625,6 +625,31 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
         }
 
         let llm_call_start = std::time::Instant::now();
+
+        // Wire up real-time token streaming to the channel layer.
+        // Bounded channel bounds memory usage when the consumer (channel
+        // layer) is slower than the LLM; producer drops chunks on overflow
+        // via `try_send`.
+        {
+            let (chunk_tx, mut chunk_rx) =
+                tokio::sync::mpsc::channel::<String>(256);
+            let channels = Arc::clone(&self.agent.channels);
+            let channel_name = self.message.channel.clone();
+            let metadata = self.message.metadata.clone();
+            tokio::spawn(async move {
+                while let Some(chunk) = chunk_rx.recv().await {
+                    let _ = channels
+                        .send_status(
+                            &channel_name,
+                            crate::channels::StatusUpdate::StreamChunk(chunk),
+                            &metadata,
+                        )
+                        .await;
+                }
+            });
+            reason_ctx.chunk_sender = Some(chunk_tx);
+        }
+
         let output = match reasoning.respond_with_tools(reason_ctx).await {
             Ok(output) => output,
             Err(crate::error::LlmError::ContextLengthExceeded { used, limit }) => {

src/llm/circuit_breaker.rs

@@ -292,6 +292,42 @@ impl LlmProvider for CircuitBreakerProvider {
         }
     }
 
+    async fn complete_stream(
+        &self,
+        request: CompletionRequest,
+        on_chunk: &mut (dyn FnMut(String) + Send),
+    ) -> Result<CompletionResponse, LlmError> {
+        self.check_allowed().await?;
+        match self.inner.complete_stream(request, on_chunk).await {
+            Ok(resp) => {
+                self.record_success().await;
+                Ok(resp)
+            }
+            Err(err) => {
+                self.record_failure(&err).await;
+                Err(err)
+            }
+        }
+    }
+
+    async fn complete_with_tools_stream(
+        &self,
+        request: ToolCompletionRequest,
+        on_chunk: &mut (dyn FnMut(String) + Send),
+    ) -> Result<ToolCompletionResponse, LlmError> {
+        self.check_allowed().await?;
+        match self.inner.complete_with_tools_stream(request, on_chunk).await {
+            Ok(resp) => {
+                self.record_success().await;
+                Ok(resp)
+            }
+            Err(err) => {
+                self.record_failure(&err).await;
+                Err(err)
+            }
+        }
+    }
+
     async fn list_models(&self) -> Result<Vec<String>, LlmError> {
         self.inner.list_models().await
     }

src/llm/failover.rs

@@ -329,6 +329,26 @@ impl LlmProvider for FailoverProvider {
         Ok(response)
     }
 
+    async fn complete_stream(
+        &self,
+        request: CompletionRequest,
+        on_chunk: &mut (dyn FnMut(String) + Send),
+    ) -> Result<CompletionResponse, LlmError> {
+        self.providers[self.last_used.load(Ordering::Relaxed)]
+            .complete_stream(request, on_chunk)
+            .await
+    }
+
+    async fn complete_with_tools_stream(
+        &self,
+        request: ToolCompletionRequest,
+        on_chunk: &mut (dyn FnMut(String) + Send),
+    ) -> Result<ToolCompletionResponse, LlmError> {
+        self.providers[self.last_used.load(Ordering::Relaxed)]
+            .complete_with_tools_stream(request, on_chunk)
+            .await
+    }
+
     fn active_model_name(&self) -> String {
         self.providers[self.last_used.load(Ordering::Relaxed)].active_model_name()
     }

src/llm/mod.rs

@@ -25,6 +25,7 @@ mod nearai_chat;
 pub mod oauth_helpers;
 pub mod openai_codex_provider;
 pub mod openai_codex_session;
+mod openai_compat_stream;
 mod provider;
 mod reasoning;
 pub mod recording;
@@ -293,7 +294,7 @@ fn create_openai_compat_from_registry(
         builder = builder.base_url(&base_url);
     }
     if !extra_headers.is_empty() {
-        builder = builder.http_headers(extra_headers);
+        builder = builder.http_headers(extra_headers.clone());
     }
 
     let client: openai::Client = builder.build().map_err(|e| LlmError::RequestFailed {
@@ -316,7 +317,41 @@ fn create_openai_compat_from_registry(
 
     let adapter = RigAdapter::new(model, &config.model)
         .with_unsupported_params(config.unsupported_params.clone());
-    Ok(Arc::new(adapter))
+    // Re-use the already-validated header map: iterate it to build the
+    // (String, String) pairs for the streaming provider, skipping any that
+    // produced warnings above.
+    let extra_headers_vec: Vec<(String, String)> = extra_headers
+        .iter()
+        .filter_map(|(name, value)| {
+            value
+                .to_str()
+                .ok()
+                .map(|v| (name.as_str().to_string(), v.to_string()))
+        })
+        .collect();
+    let unsupported: std::collections::HashSet<String> =
+        config.unsupported_params.iter().cloned().collect();
+    // Normalize the base_url the same way the rig-core client does so the
+    // streaming path hits the exact same endpoint.
+    let streaming_base_url = if config.base_url.is_empty() {
+        String::new()
+    } else {
+        normalize_openai_base_url(&config.base_url)
+    };
+    let streaming = openai_compat_stream::OpenAiCompatStreamingProvider::new(
+        Arc::new(adapter),
+        api_key,
+        streaming_base_url,
+        config.model.clone(),
+        config.provider_id.clone(),
+        extra_headers_vec,
+        unsupported,
+    )
+    .map_err(|e| LlmError::RequestFailed {
+        provider: config.provider_id.clone(),
+        reason: format!("Failed to build streaming HTTP client: {e}"),
+    })?;
+    Ok(Arc::new(streaming))
 }
 
 fn create_anthropic_from_registry(