use super::api_client::{ApiClient, AuthMethod}; use super::base::{ConfigKey, MessageStream, Provider, ProviderDef, ProviderMetadata}; use super::errors::ProviderError; use super::inventory::InventoryIdentityInput; use super::openai_compatible::handle_status; use super::retry::{ProviderRetry, RetryConfig}; use super::utils::{ImageFormat, RequestLog}; use crate::config::declarative_providers::DeclarativeProviderConfig; use crate::conversation::message::Message; use crate::model::ModelConfig; use crate::providers::formats::ollama::{create_request, response_to_streaming_message_ollama}; use anyhow::{Error, Result}; use async_stream::try_stream; use async_trait::async_trait; use futures::future::BoxFuture; use futures::TryStreamExt; use reqwest::Response; use rmcp::model::Tool; use serde_json::{json, Value}; use std::time::Duration; use tokio::pin; use tokio_stream::StreamExt; use tokio_util::codec::{FramedRead, LinesCodec}; use tokio_util::io::StreamReader; use url::Url; const OLLAMA_PROVIDER_NAME: &str = "ollama"; pub const OLLAMA_HOST: &str = "localhost"; pub const OLLAMA_TIMEOUT: u64 = 600; pub const OLLAMA_DEFAULT_PORT: u16 = 11434; pub const OLLAMA_DEFAULT_MODEL: &str = "qwen3"; pub const OLLAMA_KNOWN_MODELS: &[&str] = &[ OLLAMA_DEFAULT_MODEL, "qwen3-vl", "qwen3-coder:30b", "qwen3-coder:480b-cloud", ]; pub const OLLAMA_DOC_URL: &str = "https://ollama.com/library"; // Ollama-specific retry config: large models can take 30-120s to load into memory, // during which Ollama returns 500 errors. Use more retries with gradual backoff // to wait for the model to become ready. const OLLAMA_MAX_RETRIES: usize = 10; const OLLAMA_INITIAL_RETRY_INTERVAL_MS: u64 = 2000; const OLLAMA_BACKOFF_MULTIPLIER: f64 = 1.5; const OLLAMA_MAX_RETRY_INTERVAL_MS: u64 = 15_000; #[derive(serde::Serialize)] pub struct OllamaProvider { #[serde(skip)] api_client: ApiClient, model: ModelConfig, supports_streaming: bool, name: String, skip_canonical_filtering: bool, } fn resolve_ollama_num_ctx(model_config: &ModelConfig) -> Option { let config = crate::config::Config::global(); let input_limit = match config.get_param::("GOOSE_INPUT_LIMIT") { Ok(limit) if limit > 0 => Some(limit), Ok(_) => None, Err(crate::config::ConfigError::NotFound(_)) => None, Err(e) => { tracing::warn!("Invalid GOOSE_INPUT_LIMIT value: {}", e); None } }; input_limit.or(model_config.context_limit) } fn resolve_ollama_stream_usage() -> bool { let config = crate::config::Config::global(); match config.get_param::("OLLAMA_STREAM_USAGE") { Ok(val) => val, // Key not set: default to true. Ollama supports stream_options since // mid-2025 and most installs benefit from token usage tracking. Err(crate::config::ConfigError::NotFound(_)) => true, // Invalid value (e.g. "0", "yes", typo): warn and disable stream_options // so users who intended to opt out aren't silently left hanging. Err(e) => { tracing::warn!( "Invalid OLLAMA_STREAM_USAGE value ({}); disabling stream_options. \ Use true or false.", e ); false } } } fn apply_ollama_options(payload: &mut Value, model_config: &ModelConfig) { if let Some(obj) = payload.as_object_mut() { // Gate stream_options behind OLLAMA_STREAM_USAGE (default: true). // Older Ollama builds that don't support stream_options may stall before // emitting any SSE data, blocking until the client timeout (600s). // with_line_timeout() only protects after the first line arrives, so // users on older builds should set OLLAMA_STREAM_USAGE=false. if !resolve_ollama_stream_usage() { obj.remove("stream_options"); } // Convert max_completion_tokens / max_tokens to Ollama's options.num_predict. // Reasoning models emit max_completion_tokens; non-reasoning models emit max_tokens. let max_tokens = obj .remove("max_completion_tokens") .or_else(|| obj.remove("max_tokens")); if let Some(max_tokens) = max_tokens { let options = obj.entry("options").or_insert_with(|| json!({})); if let Some(options_obj) = options.as_object_mut() { options_obj.entry("num_predict").or_insert(max_tokens); } } // Apply num_ctx from context limit settings. if let Some(limit) = resolve_ollama_num_ctx(model_config) { let options = obj.entry("options").or_insert_with(|| json!({})); if let Some(options_obj) = options.as_object_mut() { options_obj.insert("num_ctx".to_string(), json!(limit)); } } } } fn ollama_host_configured(config: &crate::config::Config) -> bool { config.get_param::("OLLAMA_HOST").is_ok() } impl OllamaProvider { pub async fn from_env(model: ModelConfig) -> Result { let config = crate::config::Config::global(); let host: String = config .get_param("OLLAMA_HOST") .unwrap_or_else(|_| OLLAMA_HOST.to_string()); let timeout: Duration = Duration::from_secs(config.get_param("OLLAMA_TIMEOUT").unwrap_or(OLLAMA_TIMEOUT)); let base = if host.starts_with("http://") || host.starts_with("https://") { host.clone() } else { format!("http://{}", host) }; let mut base_url = Url::parse(&base).map_err(|e| anyhow::anyhow!("Invalid base URL: {e}"))?; let explicit_port = host.contains(':'); let is_localhost = host == "localhost" || host == "127.0.0.1" || host == "::1"; if base_url.port().is_none() && !explicit_port && !host.starts_with("http") && is_localhost { base_url .set_port(Some(OLLAMA_DEFAULT_PORT)) .map_err(|_| anyhow::anyhow!("Failed to set default port"))?; } let api_client = ApiClient::with_timeout(base_url.to_string(), AuthMethod::NoAuth, timeout)?; Ok(Self { api_client, model, supports_streaming: true, name: OLLAMA_PROVIDER_NAME.to_string(), skip_canonical_filtering: false, }) } pub fn from_custom_config( model: ModelConfig, config: DeclarativeProviderConfig, ) -> Result { let timeout = Duration::from_secs(config.timeout_seconds.unwrap_or(OLLAMA_TIMEOUT)); let base = if config.base_url.starts_with("http://") || config.base_url.starts_with("https://") { config.base_url.clone() } else { format!("http://{}", config.base_url) }; let mut base_url = Url::parse(&base) .map_err(|e| anyhow::anyhow!("Invalid base URL '{}': {}", config.base_url, e))?; let explicit_default_port = config.base_url.ends_with(":80") || config.base_url.ends_with(":443"); let is_https = base_url.scheme() == "https"; if base_url.port().is_none() && !explicit_default_port && !is_https { base_url .set_port(Some(OLLAMA_DEFAULT_PORT)) .map_err(|_| anyhow::anyhow!("Failed to set default port"))?; } let mut api_client = ApiClient::with_timeout(base_url.to_string(), AuthMethod::NoAuth, timeout)?; if let Some(headers) = &config.headers { let mut header_map = reqwest::header::HeaderMap::new(); for (key, value) in headers { let header_name = reqwest::header::HeaderName::from_bytes(key.as_bytes())?; let header_value = reqwest::header::HeaderValue::from_str(value)?; header_map.insert(header_name, header_value); } api_client = api_client.with_headers(header_map)?; } let supports_streaming = config.supports_streaming.unwrap_or(true); if !supports_streaming { return Err(anyhow::anyhow!( "Ollama provider does not support non-streaming mode. All Ollama models support streaming. \ Please remove 'supports_streaming: false' from your provider configuration." )); } let model = if let Some(ref fast_model_name) = config.fast_model { model.with_fast(fast_model_name, &config.name)? } else { model }; Ok(Self { api_client, model, supports_streaming, name: config.name.clone(), skip_canonical_filtering: config.skip_canonical_filtering, }) } } impl ProviderDef for OllamaProvider { type Provider = Self; fn metadata() -> ProviderMetadata { ProviderMetadata::new( OLLAMA_PROVIDER_NAME, "Ollama", "Local open source models", OLLAMA_DEFAULT_MODEL, OLLAMA_KNOWN_MODELS.to_vec(), OLLAMA_DOC_URL, vec![ ConfigKey::new("OLLAMA_HOST", true, false, Some(OLLAMA_HOST), true), ConfigKey::new( "OLLAMA_TIMEOUT", false, false, Some(&(OLLAMA_TIMEOUT.to_string())), false, ), ], ) } fn from_env( model: ModelConfig, _extensions: Vec, ) -> BoxFuture<'static, Result> { Box::pin(Self::from_env(model)) } fn supports_inventory_refresh() -> bool { true } fn inventory_configured() -> bool { ollama_host_configured(crate::config::Config::global()) } fn inventory_identity() -> Result { let config = crate::config::Config::global(); Ok( InventoryIdentityInput::new(OLLAMA_PROVIDER_NAME, OLLAMA_PROVIDER_NAME).with_public( "host", config .get_param::("OLLAMA_HOST") .unwrap_or_else(|_| OLLAMA_HOST.to_string()), ), ) } } #[async_trait] impl Provider for OllamaProvider { fn get_name(&self) -> &str { &self.name } fn skip_canonical_filtering(&self) -> bool { self.skip_canonical_filtering } fn get_model_config(&self) -> ModelConfig { self.model.clone() } fn retry_config(&self) -> RetryConfig { RetryConfig::new( OLLAMA_MAX_RETRIES, OLLAMA_INITIAL_RETRY_INTERVAL_MS, OLLAMA_BACKOFF_MULTIPLIER, OLLAMA_MAX_RETRY_INTERVAL_MS, ) .transient_only() } async fn stream( &self, model_config: &ModelConfig, session_id: &str, system: &str, messages: &[Message], tools: &[Tool], ) -> Result { let mut payload = create_request( model_config, system, messages, tools, &ImageFormat::OpenAi, true, )?; apply_ollama_options(&mut payload, model_config); let mut log = RequestLog::start(model_config, &payload)?; let response = self .with_retry(|| async { let resp = self .api_client .response_post(Some(session_id), "v1/chat/completions", &payload) .await?; handle_status(resp).await }) .await .inspect_err(|e| { let _ = log.error(e); })?; stream_ollama(response, log) } async fn fetch_supported_models(&self) -> Result, ProviderError> { let response = self .api_client .request(None, "api/tags") .response_get() .await .map_err(|e| ProviderError::RequestFailed(format!("Failed to fetch models: {}", e)))?; if !response.status().is_success() { return Err(ProviderError::RequestFailed(format!( "Failed to fetch models: HTTP {}", response.status() ))); } let json_response = response.json::().await.map_err(|e| { ProviderError::RequestFailed(format!("Failed to parse response: {}", e)) })?; let models = json_response .get("models") .and_then(|m| m.as_array()) .ok_or_else(|| { ProviderError::RequestFailed("No models array in response".to_string()) })?; let mut model_names: Vec = models .iter() .filter_map(|model| model.get("name").and_then(|n| n.as_str()).map(String::from)) .collect(); model_names.sort(); Ok(model_names) } } /// Default per-chunk timeout for Ollama streaming responses (seconds). /// Configurable via OLLAMA_STREAM_TIMEOUT, GOOSE_STREAM_TIMEOUT, or falls back /// to OLLAMA_TIMEOUT. Set high to accommodate slower models (CPU inference, /// large parameter counts, complex reasoning). const OLLAMA_DEFAULT_CHUNK_TIMEOUT_SECS: u64 = 120; /// Resolve the per-chunk stream timeout from config. /// Priority: OLLAMA_STREAM_TIMEOUT > GOOSE_STREAM_TIMEOUT > OLLAMA_TIMEOUT > default (120s). /// Zero values are treated as invalid and skipped, since a zero timeout would /// cause every chunk after the first to be treated as a stall. fn resolve_ollama_chunk_timeout() -> u64 { let config = crate::config::Config::global(); if let Ok(val) = config.get_param::("OLLAMA_STREAM_TIMEOUT") { if val > 0 { return val; } } if let Ok(val) = config.get_param::("GOOSE_STREAM_TIMEOUT") { if val > 0 { return val; } } match config.get_param::("OLLAMA_TIMEOUT") { Ok(val) if val > 0 => val, _ => OLLAMA_DEFAULT_CHUNK_TIMEOUT_SECS, } } /// Wraps a line stream with a per-item timeout at the raw SSE level. /// This detects dead connections without false-positive stalls during long /// tool-call generations where response_to_streaming_message_ollama buffers. fn with_line_timeout( stream: impl futures::Stream> + Unpin + Send + 'static, timeout_secs: u64, ) -> std::pin::Pin> + Send>> { let timeout = Duration::from_secs(timeout_secs); Box::pin(try_stream! { let mut stream = stream; // Allow time-to-first-token to be governed by the request timeout. // Only enforce per-chunk timeout after first SSE line arrives. match stream.next().await { Some(first_item) => yield first_item?, None => return, } loop { match tokio::time::timeout(timeout, stream.next()).await { Ok(Some(item)) => yield item?, Ok(None) => break, Err(_) => { Err::<(), anyhow::Error>(anyhow::anyhow!( "Ollama stream stalled: no data received for {}s. \ This may indicate the model is overwhelmed by the request payload. \ Try a smaller model or reduce the number of tools.", timeout_secs ))?; } } } }) } /// Ollama-specific streaming handler with XML tool call fallback. /// Uses the Ollama format module which buffers text when XML tool calls are detected, /// preventing duplicate content from being emitted to the UI. /// Timeout is applied at the raw SSE line level via with_line_timeout so that /// buffering inside response_to_streaming_message_ollama does not cause false stalls. fn stream_ollama(response: Response, mut log: RequestLog) -> Result { let stream = response.bytes_stream().map_err(std::io::Error::other); Ok(Box::pin(try_stream! { let stream_reader = StreamReader::new(stream); let framed = FramedRead::new(stream_reader, LinesCodec::new()) .map_err(Error::from); let chunk_timeout = resolve_ollama_chunk_timeout(); let timed_lines = with_line_timeout(framed, chunk_timeout); let message_stream = response_to_streaming_message_ollama(timed_lines); pin!(message_stream); while let Some(message) = message_stream.next().await { let (message, usage) = message.map_err(|e| ProviderError::RequestFailed(format!("Stream decode error: {}", e)) )?; log.write(&message, usage.as_ref().map(|f| f.usage).as_ref())?; yield (message, usage); } })) } #[cfg(test)] mod tests { use super::*; #[test] fn test_ollama_host_default_does_not_mark_inventory_configured() { let _guard = env_lock::lock_env([("OLLAMA_HOST", None::<&str>)]); let config_file = tempfile::NamedTempFile::new().unwrap(); let secrets_file = tempfile::NamedTempFile::new().unwrap(); let config = crate::config::Config::new_with_config_paths( vec![config_file.path().to_path_buf()], secrets_file.path(), ) .unwrap(); assert!(!ollama_host_configured(&config)); } #[test] fn test_ollama_host_env_marks_inventory_configured() { let _guard = env_lock::lock_env([("OLLAMA_HOST", Some("http://127.0.0.1:11435"))]); let config_file = tempfile::NamedTempFile::new().unwrap(); let secrets_file = tempfile::NamedTempFile::new().unwrap(); let config = crate::config::Config::new_with_config_paths( vec![config_file.path().to_path_buf()], secrets_file.path(), ) .unwrap(); assert!(ollama_host_configured(&config)); } #[test] fn test_ollama_host_config_marks_inventory_configured() { let _guard = env_lock::lock_env([("OLLAMA_HOST", None::<&str>)]); let config_file = tempfile::NamedTempFile::new().unwrap(); let secrets_file = tempfile::NamedTempFile::new().unwrap(); let config = crate::config::Config::new_with_config_paths( vec![config_file.path().to_path_buf()], secrets_file.path(), ) .unwrap(); config .set_param("OLLAMA_HOST", "http://127.0.0.1:11435") .unwrap(); assert!(ollama_host_configured(&config)); } #[test] fn test_apply_ollama_options_uses_input_limit() { let _guard = env_lock::lock_env([("GOOSE_INPUT_LIMIT", Some("8192"))]); let model_config = ModelConfig::new("qwen3") .unwrap() .with_context_limit(Some(16_000)); let mut payload = json!({}); apply_ollama_options(&mut payload, &model_config); assert_eq!(payload["options"]["num_ctx"], 8192); } #[test] fn test_apply_ollama_options_falls_back_to_context_limit() { let _guard = env_lock::lock_env([("GOOSE_INPUT_LIMIT", None::<&str>)]); let model_config = ModelConfig::new("qwen3") .unwrap() .with_context_limit(Some(12_000)); let mut payload = json!({}); apply_ollama_options(&mut payload, &model_config); assert_eq!(payload["options"]["num_ctx"], 12_000); } #[test] fn test_apply_ollama_options_skips_when_no_limit() { let _guard = env_lock::lock_env([("GOOSE_INPUT_LIMIT", None::<&str>)]); let mut model_config = ModelConfig::new("qwen3").unwrap(); model_config.context_limit = None; let mut payload = json!({}); apply_ollama_options(&mut payload, &model_config); assert!(payload.get("options").is_none()); } #[test] fn test_raw_create_request_contains_unsupported_ollama_fields() { use crate::providers::formats::ollama::create_request; use crate::providers::utils::ImageFormat; let model_config = ModelConfig::new("llama3.1") .unwrap() .with_max_tokens(Some(4096)); let messages = vec![crate::conversation::message::Message::user().with_text("hi")]; let payload = create_request( &model_config, "You are a helpful assistant.", &messages, &[], &ImageFormat::OpenAi, true, ) .unwrap(); assert!( payload.get("stream_options").is_some(), "create_request should produce stream_options for usage tracking" ); assert!( payload.get("max_tokens").is_some(), "create_request should produce max_tokens (unsupported by Ollama)" ); } #[test] fn test_apply_ollama_options_preserves_stream_options_by_default() { use crate::providers::formats::ollama::create_request; use crate::providers::utils::ImageFormat; let _guard = env_lock::lock_env([ ("GOOSE_INPUT_LIMIT", None::<&str>), ("OLLAMA_STREAM_USAGE", None::<&str>), ]); let model_config = ModelConfig::new("llama3.1") .unwrap() .with_max_tokens(Some(4096)); let messages = vec![crate::conversation::message::Message::user().with_text("hi")]; let mut payload = create_request( &model_config, "You are a helpful assistant.", &messages, &[], &ImageFormat::OpenAi, true, ) .unwrap(); apply_ollama_options(&mut payload, &model_config); assert!( payload.get("stream_options").is_some(), "stream_options should be preserved by default for usage tracking" ); assert!( payload.get("max_tokens").is_none(), "max_tokens should be removed for Ollama" ); assert!( payload.get("max_completion_tokens").is_none(), "max_completion_tokens should be removed for Ollama" ); assert_eq!( payload["options"]["num_predict"], 4096, "max_tokens should be moved to options.num_predict" ); assert_eq!(payload["stream"], true, "stream field should be preserved"); } #[test] fn test_apply_ollama_options_strips_stream_options_when_disabled() { use crate::providers::formats::ollama::create_request; use crate::providers::utils::ImageFormat; let _guard = env_lock::lock_env([ ("GOOSE_INPUT_LIMIT", None::<&str>), ("OLLAMA_STREAM_USAGE", Some("false")), ]); let model_config = ModelConfig::new("llama3.1") .unwrap() .with_max_tokens(Some(4096)); let messages = vec![crate::conversation::message::Message::user().with_text("hi")]; let mut payload = create_request( &model_config, "You are a helpful assistant.", &messages, &[], &ImageFormat::OpenAi, true, ) .unwrap(); apply_ollama_options(&mut payload, &model_config); assert!( payload.get("stream_options").is_none(), "stream_options should be removed when OLLAMA_STREAM_USAGE=false" ); } #[test] fn test_resolve_ollama_chunk_timeout_defaults_to_ollama_timeout() { let _guard = env_lock::lock_env([ ("OLLAMA_STREAM_TIMEOUT", None::<&str>), ("GOOSE_STREAM_TIMEOUT", None::<&str>), ("OLLAMA_TIMEOUT", Some("300")), ]); assert_eq!(resolve_ollama_chunk_timeout(), 300); } #[test] fn test_resolve_ollama_chunk_timeout_prefers_stream_override() { let _guard = env_lock::lock_env([ ("OLLAMA_STREAM_TIMEOUT", Some("60")), ("GOOSE_STREAM_TIMEOUT", Some("90")), ("OLLAMA_TIMEOUT", Some("300")), ]); assert_eq!(resolve_ollama_chunk_timeout(), 60); } #[test] fn test_resolve_ollama_chunk_timeout_uses_goose_stream_fallback() { let _guard = env_lock::lock_env([ ("OLLAMA_STREAM_TIMEOUT", None::<&str>), ("GOOSE_STREAM_TIMEOUT", Some("90")), ("OLLAMA_TIMEOUT", Some("300")), ]); assert_eq!(resolve_ollama_chunk_timeout(), 90); } #[test] fn test_resolve_ollama_chunk_timeout_uses_default_when_unset() { let _guard = env_lock::lock_env([ ("OLLAMA_STREAM_TIMEOUT", None::<&str>), ("GOOSE_STREAM_TIMEOUT", None::<&str>), ("OLLAMA_TIMEOUT", None::<&str>), ]); assert_eq!( resolve_ollama_chunk_timeout(), OLLAMA_DEFAULT_CHUNK_TIMEOUT_SECS ); } #[test] fn test_resolve_ollama_chunk_timeout_skips_zero_values() { let _guard = env_lock::lock_env([ ("OLLAMA_STREAM_TIMEOUT", Some("0")), ("GOOSE_STREAM_TIMEOUT", Some("0")), ("OLLAMA_TIMEOUT", Some("300")), ]); assert_eq!(resolve_ollama_chunk_timeout(), 300); } #[test] fn test_resolve_ollama_chunk_timeout_skips_all_zero_to_default() { let _guard = env_lock::lock_env([ ("OLLAMA_STREAM_TIMEOUT", Some("0")), ("GOOSE_STREAM_TIMEOUT", Some("0")), ("OLLAMA_TIMEOUT", Some("0")), ]); assert_eq!( resolve_ollama_chunk_timeout(), OLLAMA_DEFAULT_CHUNK_TIMEOUT_SECS ); } #[test] fn test_ollama_retry_config_is_transient_only() { let config = RetryConfig::new( OLLAMA_MAX_RETRIES, OLLAMA_INITIAL_RETRY_INTERVAL_MS, OLLAMA_BACKOFF_MULTIPLIER, OLLAMA_MAX_RETRY_INTERVAL_MS, ) .transient_only(); assert!(config.transient_only); use super::super::errors::ProviderError; use super::super::retry::should_retry; assert!(!should_retry( &ProviderError::RequestFailed("Resource not found (404)".into()), &config )); assert!(!should_retry( &ProviderError::RequestFailed("Bad request (400)".into()), &config )); assert!(should_retry( &ProviderError::ServerError("500 model loading".into()), &config )); assert!(should_retry( &ProviderError::NetworkError("connection refused".into()), &config )); } }