Merge branch 'main' into feat/plugin

server/src/app.rs

@@ -44,6 +44,7 @@ impl App {
             store.clone(),
             plugins.clone(),
             config.provider_request_timeout,
+            config.provider_stream_idle_timeout,
         ));
         let registry = TransportRegistry::with_plugins(
             store.clone(),

server/src/config.rs

@@ -10,7 +10,8 @@ const DATA_DIR_NAME: &str = ".cursor-byok-v3";
 const DATABASE_FILE_NAME: &str = "cursor-byok.db";
 const V0049_DATA_DIR_NAME: &str = ".cursor-local-assistant-v2";
 const V0049_CONFIG_FILE_NAME: &str = "config.yaml";
-const DEFAULT_PROVIDER_REQUEST_TIMEOUT: Duration = Duration::from_secs(3000);
+const DEFAULT_PROVIDER_REQUEST_TIMEOUT: Duration = Duration::from_secs(60 * 60);
+const DEFAULT_PROVIDER_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(30 * 60);
 
 pub fn managed_data_dir() -> Result<PathBuf> {
     let home_dir = dirs::home_dir()
@@ -54,6 +55,7 @@ pub struct Config {
     pub listen_addr: SocketAddr,
     pub database_url: String,
     pub provider_request_timeout: Duration,
+    pub provider_stream_idle_timeout: Duration,
     pub console: Option<ConsoleSource>,
     pub use_persisted_ports: bool,
 }
@@ -104,6 +106,7 @@ impl Config {
             listen_addr,
             database_url: database_url_from_env()?,
             provider_request_timeout: request_timeout,
+            provider_stream_idle_timeout: DEFAULT_PROVIDER_STREAM_IDLE_TIMEOUT,
             console,
             use_persisted_ports: false,
         })
@@ -116,6 +119,7 @@ impl Config {
                 .expect("desktop listen address is static"),
             database_url: default_database_url()?,
             provider_request_timeout: DEFAULT_PROVIDER_REQUEST_TIMEOUT,
+            provider_stream_idle_timeout: DEFAULT_PROVIDER_STREAM_IDLE_TIMEOUT,
             console: None,
             use_persisted_ports: true,
         })
@@ -144,3 +148,20 @@ fn database_url_for_dir(data_dir: &std::path::Path) -> Result<String> {
         .ok_or_else(|| Error::Config("database path is not valid UTF-8".into()))?;
     Ok(format!("sqlite://{database_path}"))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn provider_timeout_defaults_match_runtime_boundaries() {
+        assert_eq!(
+            DEFAULT_PROVIDER_STREAM_IDLE_TIMEOUT,
+            Duration::from_secs(30 * 60)
+        );
+        assert_eq!(
+            DEFAULT_PROVIDER_REQUEST_TIMEOUT,
+            Duration::from_secs(60 * 60)
+        );
+    }
+}

server/src/provider/anthropic.rs

@@ -12,7 +12,7 @@ use crate::{
 };
 
 use super::{
-    merge_extra_params,
+    map_sse_error, merge_extra_params, provider_event_error,
     recorder::recorded_headers,
     retry::{send_with_retry, Attempt, RetryPolicy},
     CallRecorder, FinishReason, ModelEvent, Provider, ProviderStream,
@@ -129,8 +129,11 @@ impl Provider for AnthropicProvider {
                 _ = cancellation.cancelled() => { return; }
                 event = source.next() => event,
             } {
-                let event = event.map_err(|error| Error::Provider(format!("Anthropic SSE: {error}")))?;
+                let event = event.map_err(|error| map_sse_error("Anthropic", error))?;
                 let value: Value = serde_json::from_str(&event.data)?;
+                if let Some(error) = provider_event_error("Anthropic", &value) {
+                    Err(error)?;
+                }
                 let data_kind = value.get("type").and_then(Value::as_str);
                 let kind = match event.event.as_str() {
                     "" | "message" => data_kind.unwrap_or(event.event.as_str()),
@@ -242,7 +245,6 @@ impl Provider for AnthropicProvider {
                         };
                         yield ModelEvent::Done(finish);
                     }
-                    "error" => Err(Error::Provider(format!("Anthropic stream error: {}", event.data)))?,
                     _ => {}
                 }
             }

server/src/provider/mod.rs

@@ -32,6 +32,123 @@ pub trait Provider: Send + Sync {
     ) -> ProviderStream;
 }
 
+fn map_sse_error(
+    label: &str,
+    error: eventsource_stream::EventStreamError<crate::Error>,
+) -> crate::Error {
+    match error {
+        eventsource_stream::EventStreamError::Transport(error) => error,
+        eventsource_stream::EventStreamError::Utf8(error) => {
+            crate::Error::Provider(format!("{label} SSE UTF-8 error: {error}"))
+        }
+        eventsource_stream::EventStreamError::Parser(error) => {
+            crate::Error::Provider(format!("{label} SSE parse error: {error}"))
+        }
+    }
+}
+
+fn provider_event_error(label: &str, value: &serde_json::Value) -> Option<crate::Error> {
+    let kind = value.get("type").and_then(serde_json::Value::as_str);
+    let direct_error = value.get("error").filter(|error| !error.is_null());
+    if !matches!(kind, Some("error" | "response.failed")) && direct_error.is_none() {
+        return None;
+    }
+
+    let message = value
+        .get("message")
+        .and_then(serde_json::Value::as_str)
+        .or_else(|| {
+            value
+                .pointer("/error/message")
+                .and_then(serde_json::Value::as_str)
+        })
+        .or_else(|| {
+            value
+                .pointer("/response/error/message")
+                .and_then(serde_json::Value::as_str)
+        })
+        .or_else(|| direct_error.and_then(serde_json::Value::as_str))
+        .or_else(|| {
+            value
+                .pointer("/response/error")
+                .and_then(serde_json::Value::as_str)
+        })
+        .unwrap_or("provider returned an error event without a message");
+
+    Some(crate::Error::Provider(format!("{label} error: {message}")))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn sse_transport_errors_are_not_relabelled_as_parse_errors() {
+        let error = map_sse_error(
+            "test provider",
+            eventsource_stream::EventStreamError::Transport(crate::Error::Provider(
+                "connection closed".into(),
+            )),
+        );
+
+        let crate::Error::Provider(message) = error else {
+            panic!("transport error category must be preserved");
+        };
+        assert_eq!(message, "connection closed");
+    }
+
+    #[test]
+    fn provider_error_events_extract_flat_and_nested_messages() {
+        assert_provider_error(
+            "OpenAI Responses",
+            serde_json::json!({
+                "type": "error",
+                "message": "Internal error during token generation"
+            }),
+            "OpenAI Responses error: Internal error during token generation",
+        );
+        assert_provider_error(
+            "OpenAI Chat",
+            serde_json::json!({
+                "error": {"message": "quota exceeded", "type": "server_error"}
+            }),
+            "OpenAI Chat error: quota exceeded",
+        );
+        assert_provider_error(
+            "Anthropic",
+            serde_json::json!({
+                "type": "error",
+                "error": {"type": "overloaded_error", "message": "Overloaded"}
+            }),
+            "Anthropic error: Overloaded",
+        );
+        assert_provider_error(
+            "OpenAI Responses",
+            serde_json::json!({
+                "type": "response.failed",
+                "response": {"error": {"message": "generation failed"}}
+            }),
+            "OpenAI Responses error: generation failed",
+        );
+    }
+
+    #[test]
+    fn successful_provider_events_are_not_errors() {
+        assert!(provider_event_error(
+            "OpenAI Responses",
+            &serde_json::json!({"type": "response.completed", "error": null})
+        )
+        .is_none());
+    }
+
+    fn assert_provider_error(label: &str, value: serde_json::Value, expected: &str) {
+        let Some(crate::Error::Provider(message)) = provider_event_error(label, &value) else {
+            panic!("expected provider error");
+        };
+        assert_eq!(message, expected);
+    }
+}
+
 fn merge_extra_params(body: &mut serde_json::Value, extra: &serde_json::Value) -> Result<()> {
     let extra = extra
         .as_object()

server/src/provider/openai_chat.rs

@@ -17,7 +17,11 @@ use crate::{
 };
 
 use super::{
+<<<<<<< HEAD
     apply_body_allowlist, apply_openai_prompt_cache_key, merge_extra_params,
+=======
+    apply_openai_prompt_cache_key, map_sse_error, merge_extra_params, provider_event_error,
+>>>>>>> main
     recorder::recorded_headers,
     retry::{send_with_retry, Attempt, RetryPolicy},
     CallRecorder, FinishReason, ModelEvent, Provider, ProviderStream,
@@ -147,12 +151,14 @@ impl Provider for OpenAiChatProvider {
                     break;
                 };
                 let event = event.map_err(|error| {
-                    let err_msg = error.to_string();
                     tracing::debug!(iteration = loop_iteration, error = %error, "OpenAI Chat SSE event failed");
-                    Error::Provider(format!("OpenAI Chat SSE: {err_msg}"))
+                    map_sse_error("OpenAI Chat", error)
                 })?;
                 if event.data == "[DONE]" { saw_done_marker = true; break; }
                 let value: Value = serde_json::from_str(&event.data)?;
+                if let Some(error) = provider_event_error("OpenAI Chat", &value) {
+                    Err(error)?;
+                }
                 if let Some(usage) = value.get("usage").filter(|value| !value.is_null()) {
                     final_usage = Some(openai_usage(usage));
                 }

server/src/provider/openai_responses.rs

@@ -14,7 +14,11 @@ use crate::{
 };
 
 use super::{
+<<<<<<< HEAD
     apply_body_allowlist, apply_openai_prompt_cache_key, merge_extra_params,
+=======
+    apply_openai_prompt_cache_key, map_sse_error, merge_extra_params, provider_event_error,
+>>>>>>> main
     recorder::recorded_headers,
     retry::{send_with_retry, Attempt, RetryPolicy},
     CallRecorder, FinishReason, ModelEvent, Provider, ProviderStream,
@@ -127,9 +131,12 @@ impl Provider for OpenAiResponsesProvider {
                     event = source.next() => event,
                 };
                 let Some(event) = event else { break };
-                let event = event.map_err(|error| Error::Provider(format!("OpenAI Responses SSE: {error}")))?;
+                let event = event.map_err(|error| map_sse_error("OpenAI Responses", error))?;
                 if event.data == "[DONE]" { break; }
                 let value: Value = serde_json::from_str(&event.data)?;
+                if let Some(error) = provider_event_error("OpenAI Responses", &value) {
+                    Err(error)?;
+                }
                 let kind = value.get("type").and_then(Value::as_str).unwrap_or(&event.event);
                 match kind {
                     "response.output_text.delta" => {
@@ -248,7 +255,6 @@ impl Provider for OpenAiResponsesProvider {
                         terminal = true;
                         yield ModelEvent::Done(FinishReason::Length);
                     }
-                    "response.failed" => Err(Error::Provider(format!("OpenAI Responses failed: {}", event.data)))?,
                     _ => {}
                 }
             }

server/src/provider/router.rs

@@ -24,14 +24,20 @@ pub struct ProviderRouter {
     store: Store,
     plugins: PluginRegistry,
     request_timeout: Duration,
+    stream_idle_timeout: Duration,
 }
 
 impl ProviderRouter {
+<<<<<<< HEAD
     pub fn new(store: Store, plugins: PluginRegistry, request_timeout: Duration) -> Self {
+=======
+    pub fn new(store: Store, request_timeout: Duration, stream_idle_timeout: Duration) -> Self {
+>>>>>>> main
         Self {
             store,
             plugins,
             request_timeout,
+            stream_idle_timeout,
         }
     }
 }
@@ -45,8 +51,10 @@ impl Provider for ProviderRouter {
         let store = self.store.clone();
         let plugins = self.plugins.clone();
         let request_timeout = self.request_timeout;
+        let stream_idle_timeout = self.stream_idle_timeout;
         Box::pin(try_stream! {
             let selected = invocation.request.model.model_id.clone();
+<<<<<<< HEAD
             if selected.starts_with(ADAPTER_ID_PREFIX) {
                 // 插件模型与内置模型走完全相同的流程:Recorder、统一事件、
                 // 规范化包装。资源选择与将来的负载均衡都在插件 Provider 内部。
@@ -70,6 +78,131 @@ impl Provider for ProviderRouter {
                     match item {
                         Ok(event) => { recorder.event(&event).await?; yield event; }
                         Err(error) => { recorder.failed(&error).await?; Err(error)?; }
+=======
+            let model = store
+                .model(&selected)
+                .await?
+                .ok_or_else(|| Error::Provider(format!("unknown model: {selected}")))?;
+            let provider_type = model.provider_type();
+            let request_url = model.request_url()?;
+            model.configure(&mut invocation.request.model);
+            invocation.request.model.extra_params = model.extra_params().clone();
+            invocation.request.model.model_id = model.model_id.clone();
+            let recorder = CallRecorder::start(store.clone(), NewLlmCall {
+                call_id: invocation.call_id.clone(),
+                run_id: invocation.run_id.clone(),
+                conversation_id: invocation.conversation_id.clone(),
+                provider_call_index: invocation.provider_call_index.min(i64::MAX as u64) as i64,
+                model_hash: model.model_hash.clone(),
+                provider_type,
+                provider_url: model.base_url.clone(),
+                request_type: provider_type,
+                request_url: request_url.clone(),
+                model_id: model.model_id.clone(),
+                display_name: model.display_name.clone(),
+                reasoning_effort: invocation.request.model.reasoning.effort.clone(),
+                fast: invocation.request.model.latency == ModelLatency::Fast,
+                message_count: invocation.request.history.len(),
+                tool_count: invocation.request.prompt.tools.len(),
+                detailed: false,
+            }).await?;
+            let _cancel_on_drop = recorder.cancel_on_drop();
+            let config = ProviderConfig {
+                kind: match provider_type {
+                    ProviderType::OpenAiChat => ProviderKind::OpenAiChat,
+                    ProviderType::OpenAiResponses => ProviderKind::OpenAiResponses,
+                    ProviderType::Anthropic => ProviderKind::Anthropic,
+                },
+                request_url,
+                api_key: model.api_key.clone(),
+                custom_headers: if model.custom_headers_enabled {
+                    custom_headers(&model.custom_headers)?
+                } else {
+                    reqwest::header::HeaderMap::new()
+                },
+                max_output_tokens: model.max_output_tokens(),
+                request_timeout,
+            };
+            let client = crate::network::client_builder(&store)
+                .await?
+                .timeout(config.request_timeout)
+                .build()?;
+            let provider = build_observed(&config, recorder.clone(), client)?;
+            let stream_cancellation = cancellation.clone();
+            let mut stream = provider.stream(invocation, cancellation);
+            let stream_started = std::time::Instant::now();
+            tracing::debug!(
+                model = %selected,
+                provider_type = ?provider_type,
+                request_timeout_ms = config.request_timeout.as_millis() as u64,
+                stream_idle_timeout_ms = stream_idle_timeout.as_millis() as u64,
+                "provider stream created"
+            );
+            let mut last_event_time = std::time::Instant::now();
+            let mut event_count: u64 = 0;
+            loop {
+                let event = match next_provider_event(&mut stream, stream_idle_timeout).await {
+                    Ok(Some(event)) => event,
+                    Ok(None) => break,
+                    Err(_) => {
+                        let elapsed_ms = stream_started.elapsed().as_millis() as u64;
+                        let error = stream_idle_timeout_error(stream_idle_timeout);
+                        tracing::warn!(
+                            error = %error,
+                            elapsed_ms,
+                            event_count,
+                            idle_timeout_ms = stream_idle_timeout.as_millis() as u64,
+                            "provider stream idle timeout"
+                        );
+                        Err(error)
+                    }
+                };
+                let now = std::time::Instant::now();
+                let gap_ms = now.duration_since(last_event_time).as_millis() as u64;
+                let elapsed_ms = now.duration_since(stream_started).as_millis() as u64;
+                event_count += 1;
+                match event {
+                    Ok(event) => {
+                        let event_name = match &event {
+                            super::ModelEvent::Start { .. } => "Start",
+                            super::ModelEvent::TextStart => "TextStart",
+                            super::ModelEvent::TextDelta(_) => "TextDelta",
+                            super::ModelEvent::TextEnd => "TextEnd",
+                            super::ModelEvent::ThinkingStart => "ThinkingStart",
+                            super::ModelEvent::ThinkingDelta(_) => "ThinkingDelta",
+                            super::ModelEvent::ThinkingEnd => "ThinkingEnd",
+                            super::ModelEvent::ToolCallStart { .. } => "ToolCallStart",
+                            super::ModelEvent::ToolCallArgumentsDelta { .. } => "ToolCallArgsDelta",
+                            super::ModelEvent::ToolCallEnd { .. } => "ToolCallEnd",
+                            super::ModelEvent::ProviderReplayState(_) => "ReplayState",
+                            super::ModelEvent::Usage(_) => "Usage",
+                            super::ModelEvent::Done(_) => "Done",
+                        };
+                        if gap_ms > 5000 {
+                            tracing::debug!(
+                                gap_ms,
+                                elapsed_ms,
+                                event = event_name,
+                                event_count,
+                                "slow gap detected between provider events"
+                            );
+                        }
+                        recorder.event(&event).await?;
+                        last_event_time = now;
+                        yield event;
+                    }
+                    Err(error) => {
+                        let error = normalize_provider_stream_error(error, request_timeout);
+                        tracing::debug!(
+                            error = %error,
+                            elapsed_ms,
+                            gap_ms,
+                            event_count,
+                            "provider stream error"
+                        );
+                        recorder.failed(&error).await?;
+                        Err(error)?;
+>>>>>>> main
                     }
                 }
                 finish_stream(&recorder, &cancellation).await?;
@@ -108,6 +241,7 @@ impl Provider for ProviderRouter {
     }
 }
 
+<<<<<<< HEAD
 async fn start_recorder(
     store: &Store,
     invocation: &ModelInvocation,
@@ -177,6 +311,48 @@ fn provider_kind(provider_type: ProviderType) -> ProviderKind {
         // 内置模型的 provider_type 只来自 ModelType,不可能是插件。
         ProviderType::Plugin => unreachable!("plugin models never use built-in provider configs"),
     }
+=======
+async fn next_provider_event(
+    stream: &mut ProviderStream,
+    idle_timeout: Duration,
+) -> std::result::Result<Option<Result<super::ModelEvent>>, tokio::time::error::Elapsed> {
+    tokio::time::timeout(idle_timeout, stream.next()).await
+}
+
+fn stream_idle_timeout_error(idle_timeout: Duration) -> Error {
+    Error::Provider(format!(
+        "provider stream idle timeout: no events received for {} seconds ({} minutes)",
+        idle_timeout.as_secs(),
+        idle_timeout.as_secs() / 60
+    ))
+}
+
+fn request_timeout_error(request_timeout: Duration) -> Error {
+    Error::Provider(format!(
+        "provider request timed out after {} seconds ({} minutes)",
+        request_timeout.as_secs(),
+        request_timeout.as_secs() / 60
+    ))
+}
+
+fn normalize_provider_stream_error(error: Error, request_timeout: Duration) -> Error {
+    match error {
+        Error::Http(source) if source.is_timeout() => request_timeout_error(request_timeout),
+        Error::Http(source) if source.is_body() => Error::Provider(format!(
+            "provider stream transport failed while reading the response body: {}",
+            root_error_message(&source)
+        )),
+        error => error,
+    }
+}
+
+fn root_error_message(error: &(dyn std::error::Error + 'static)) -> String {
+    let mut current = error;
+    while let Some(source) = current.source() {
+        current = source;
+    }
+    current.to_string()
+>>>>>>> main
 }
 
 fn custom_headers(value: &serde_json::Value) -> Result<reqwest::header::HeaderMap> {
@@ -233,3 +409,36 @@ fn build_inner(
     };
     Ok(Arc::new(NormalizedProvider::new(provider)))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn pending_provider_event_hits_the_idle_timeout() {
+        let mut stream: ProviderStream = Box::pin(futures_util::stream::pending());
+
+        let result = next_provider_event(&mut stream, Duration::from_millis(1)).await;
+
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn timeout_errors_state_the_boundary_and_duration() {
+        let Error::Provider(idle) = stream_idle_timeout_error(Duration::from_secs(30 * 60)) else {
+            panic!("idle timeout must be a provider error");
+        };
+        assert_eq!(
+            idle,
+            "provider stream idle timeout: no events received for 1800 seconds (30 minutes)"
+        );
+
+        let Error::Provider(request) = request_timeout_error(Duration::from_secs(60 * 60)) else {
+            panic!("request timeout must be a provider error");
+        };
+        assert_eq!(
+            request,
+            "provider request timed out after 3600 seconds (60 minutes)"
+        );
+    }
+}