feat(installer): align installer api test with bitfun-core

BitFun-Installer/src-tauri/Cargo.toml

@@ -23,15 +23,19 @@ tauri-plugin-dialog = "2"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 tokio = { version = "1", features = ["full"] }
+tokio-stream = "0.1"
 anyhow = "1.0"
 log = "0.4"
 dirs = "5.0"
 zip = "0.6"
 flate2 = "1.0"
 tar = "0.4"
 chrono = "0.4"
-reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
+reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "stream"] }
 urlencoding = "2"
+futures = "0.3"
+eventsource-stream = "0.2"
+installer-ai-stream = { path = "crates/installer-ai-stream" }
 
 [target.'cfg(windows)'.dependencies]
 winreg = "0.52"

BitFun-Installer/src-tauri/crates/installer-ai-stream/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "installer-ai-stream"
+version = "0.2.1"
+edition = "2021"
+
+[lib]
+path = "lib.rs"
+
+[dependencies]
+anyhow = "1.0"
+chrono = "0.4"
+eventsource-stream = "0.2"
+futures = "0.3"
+reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "stream"] }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "io-util", "sync", "time"] }
+log = "0.4"

BitFun-Installer/src-tauri/crates/installer-ai-stream/lib.rs

@@ -0,0 +1,8 @@
+mod stream_handler;
+mod types;
+
+pub use stream_handler::handle_anthropic_stream;
+pub use stream_handler::handle_gemini_stream;
+pub use stream_handler::handle_openai_stream;
+pub use stream_handler::handle_responses_stream;
+pub use types::unified::{UnifiedResponse, UnifiedTokenUsage, UnifiedToolCall};

BitFun-Installer/src-tauri/crates/installer-ai-stream/stream_handler/anthropic.rs

@@ -0,0 +1,172 @@
+use super::stream_stats::StreamStats;
+use crate::types::anthropic::{
+    AnthropicSSEError, ContentBlock, ContentBlockDelta, ContentBlockStart, MessageDelta,
+    MessageStart, Usage,
+};
+use crate::types::unified::UnifiedResponse;
+use anyhow::{anyhow, Result};
+use eventsource_stream::Eventsource;
+use futures::StreamExt;
+use log::{error, trace};
+use reqwest::Response;
+use std::time::Duration;
+use tokio::sync::mpsc;
+use tokio::time::timeout;
+
+/// Convert a byte stream into a structured response stream
+///
+/// # Arguments
+/// * `response` - HTTP response
+/// * `tx_event` - parsed event sender
+/// * `tx_raw_sse` - optional raw SSE sender (collect raw data for diagnostics)
+pub async fn handle_anthropic_stream(
+    response: Response,
+    tx_event: mpsc::UnboundedSender<Result<UnifiedResponse>>,
+    tx_raw_sse: Option<mpsc::UnboundedSender<String>>,
+) {
+    let mut stream = response.bytes_stream().eventsource();
+    let idle_timeout = Duration::from_secs(600);
+    let mut usage = Usage::default();
+    let mut stats = StreamStats::new("Anthropic");
+
+    loop {
+        let sse_event = timeout(idle_timeout, stream.next()).await;
+        let sse = match sse_event {
+            Ok(Some(Ok(sse))) => sse,
+            Ok(None) => {
+                let error_msg = "SSE Error: stream closed before response completed";
+                stats.log_summary("stream_closed_before_completion");
+                error!("{}", error_msg);
+                let _ = tx_event.send(Err(anyhow!(error_msg)));
+                return;
+            }
+            Ok(Some(Err(e))) => {
+                let error_msg = format!("SSE Error: {}", e);
+                stats.log_summary("sse_stream_error");
+                error!("{}", error_msg);
+                let _ = tx_event.send(Err(anyhow!(error_msg)));
+                return;
+            }
+            Err(_) => {
+                let error_msg = "SSE Timeout: idle timeout waiting for SSE";
+                stats.log_summary("sse_stream_timeout");
+                error!("{}", error_msg);
+                let _ = tx_event.send(Err(anyhow!(error_msg)));
+                return;
+            }
+        };
+
+        trace!("Anthropic SSE: {:?}", sse);
+        let event_type = sse.event;
+        let data = sse.data;
+        stats.record_sse_event(&event_type);
+
+        if let Some(ref tx) = tx_raw_sse {
+            let _ = tx.send(format!("[{}] {}", event_type, data));
+        }
+
+        match event_type.as_str() {
+            "message_start" => {
+                let message_start: MessageStart = match serde_json::from_str(&data) {
+                    Ok(message_start) => message_start,
+                    Err(e) => {
+                        stats.increment("error:sse_parsing");
+                        let err_str = format!("SSE Parsing Error: {e}, data: {}", &data);
+                        error!("{}", err_str);
+                        continue;
+                    }
+                };
+                if let Some(message_usage) = message_start.message.usage {
+                    usage.update(&message_usage);
+                }
+            }
+            "content_block_start" => {
+                let content_block_start: ContentBlockStart = match serde_json::from_str(&data) {
+                    Ok(content_block_start) => content_block_start,
+                    Err(e) => {
+                        stats.increment("error:sse_parsing");
+                        let err_str = format!("SSE Parsing Error: {e}, data: {}", &data);
+                        error!("{}", err_str);
+                        continue;
+                    }
+                };
+                if matches!(
+                    content_block_start.content_block,
+                    ContentBlock::ToolUse { .. }
+                ) {
+                    let unified_response = UnifiedResponse::from(content_block_start);
+                    trace!("Anthropic unified response: {:?}", unified_response);
+                    stats.record_unified_response(&unified_response);
+                    let _ = tx_event.send(Ok(unified_response));
+                }
+            }
+            "content_block_delta" => {
+                let content_block_delta: ContentBlockDelta = match serde_json::from_str(&data) {
+                    Ok(content_block_delta) => content_block_delta,
+                    Err(e) => {
+                        stats.increment("error:sse_parsing");
+                        let err_str = format!("SSE Parsing Error: {e}, data: {}", &data);
+                        error!("{}", err_str);
+                        continue;
+                    }
+                };
+                match UnifiedResponse::try_from(content_block_delta) {
+                    Ok(unified_response) => {
+                        trace!("Anthropic unified response: {:?}", unified_response);
+                        stats.record_unified_response(&unified_response);
+                        let _ = tx_event.send(Ok(unified_response));
+                    }
+                    Err(e) => {
+                        stats.increment("skip:invalid_content_block_delta");
+                        error!("Skipping invalid content_block_delta: {}", e);
+                    }
+                };
+            }
+            "message_delta" => {
+                let mut message_delta: MessageDelta = match serde_json::from_str(&data) {
+                    Ok(message_delta) => message_delta,
+                    Err(e) => {
+                        stats.increment("error:sse_parsing");
+                        let err_str = format!("SSE Parsing Error: {e}, data: {}", &data);
+                        error!("{}", err_str);
+                        continue;
+                    }
+                };
+                if let Some(delta_usage) = message_delta.usage.as_ref() {
+                    usage.update(delta_usage);
+                }
+                message_delta.usage = if usage.is_empty() {
+                    None
+                } else {
+                    Some(usage.clone())
+                };
+                let unified_response = UnifiedResponse::from(message_delta);
+                trace!("Anthropic unified response: {:?}", unified_response);
+                stats.record_unified_response(&unified_response);
+                let _ = tx_event.send(Ok(unified_response));
+            }
+            "error" => {
+                let sse_error: AnthropicSSEError = match serde_json::from_str(&data) {
+                    Ok(message_delta) => message_delta,
+                    Err(e) => {
+                        stats.increment("error:sse_parsing");
+                        let err_str = format!("SSE Parsing Error: {e}, data: {}", &data);
+                        stats.log_summary("sse_parsing_error");
+                        error!("{}", err_str);
+                        let _ = tx_event.send(Err(anyhow!(err_str)));
+                        return;
+                    }
+                };
+                stats.increment("error:api");
+                stats.log_summary("error_event_received");
+                let _ = tx_event.send(Err(anyhow!(String::from(sse_error.error))));
+                return;
+            }
+            "message_stop" => {
+                stats.log_summary("message_stop");
+                return;
+            }
+            _ => {}
+        }
+    }
+}