diff --git a/server/src/app.rs b/server/src/app.rs index a7bb427..3893d2d 100644 --- a/server/src/app.rs +++ b/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(), diff --git a/server/src/config.rs b/server/src/config.rs index e0b4ae3..a9c1964 100644 --- a/server/src/config.rs +++ b/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 { 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, 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 { .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) + ); + } +} diff --git a/server/src/provider/anthropic.rs b/server/src/provider/anthropic.rs index 6badbef..c4bfd6e 100644 --- a/server/src/provider/anthropic.rs +++ b/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)))?, _ => {} } } diff --git a/server/src/provider/mod.rs b/server/src/provider/mod.rs index d217aca..d997f3f 100644 --- a/server/src/provider/mod.rs +++ b/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 { + 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 { + 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() diff --git a/server/src/provider/openai_chat.rs b/server/src/provider/openai_chat.rs index 18b5d56..77290bd 100644 --- a/server/src/provider/openai_chat.rs +++ b/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)); } diff --git a/server/src/provider/openai_responses.rs b/server/src/provider/openai_responses.rs index 75da860..03feb5f 100644 --- a/server/src/provider/openai_responses.rs +++ b/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)))?, _ => {} } } diff --git a/server/src/provider/router.rs b/server/src/provider/router.rs index f0ff42e..2be0eb4 100644 --- a/server/src/provider/router.rs +++ b/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>, 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 { @@ -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)" + ); + } +}