fix: Provider stream stalls hangs the agent loop

README.md

@@ -141,7 +141,7 @@ Use `aether-agent-core` as a Rust library to build your own agent in ~25 lines.
    #[tokio::main]
    async fn main() -> Result<(), Box<dyn std::error::Error>> {
        // 1. Create a provider (reads ANTHROPIC_API_KEY from env)
-       let llm = AnthropicProvider::new(None)?;
+       let llm = AnthropicProvider::new(None);
 
        // 2. Spawn MCP tool servers from one or more mcp.json files
        let mut mcp_runtime = mcp(".").from_json_files(&["mcp.json"])?.spawn().await?;

crates/aether-cli/src/provider_connection_args.rs

@@ -1,4 +1,5 @@
 use std::collections::BTreeMap;
+use std::num::NonZeroU64;
 use std::str::FromStr;
 
 use llm::{ProviderAuthMode, ProviderConnectionOverride, ProviderConnectionOverrides};
@@ -7,7 +8,7 @@ use llm::{ProviderAuthMode, ProviderConnectionOverride, ProviderConnectionOverri
 pub struct ProviderConnectionArgs {
     #[arg(
         long = "provider",
-        value_name = "PROVIDER.url=URL|PROVIDER.auth=default|none|PROVIDER.request-model=MODEL|bedrock.inference-profile-arn=ARN"
+        value_name = "PROVIDER.url=URL|PROVIDER.auth=default|none|PROVIDER.request-model=MODEL|PROVIDER.idle-timeout-secs=SECS|bedrock.inference-profile-arn=ARN"
     )]
     pub providers: Vec<ProviderArg>,
 }
@@ -35,7 +36,7 @@ impl FromStr for ProviderArg {
         let (key, setting) = split_key_value(value)?;
         let (provider, field) = key
             .split_once('.')
-            .ok_or_else(|| "provider override must be PROVIDER.url=URL, PROVIDER.auth=default|none, PROVIDER.request-model=MODEL, or bedrock.inference-profile-arn=ARN".to_string())?;
+            .ok_or_else(|| "provider override must be PROVIDER.url=URL, PROVIDER.auth=default|none, PROVIDER.request-model=MODEL, PROVIDER.idle-timeout-secs=SECS, or bedrock.inference-profile-arn=ARN".to_string())?;
 
         validate_provider(provider)?;
         if setting.trim().is_empty() {
@@ -49,6 +50,7 @@ impl FromStr for ProviderArg {
             }
             "auth" => ProviderConnectionOverride::auth(parse_auth_mode(setting)?),
             "request-model" => ProviderConnectionOverride::request_model(setting),
+            "idle-timeout-secs" => ProviderConnectionOverride::idle_timeout_secs(parse_idle_timeout_secs(setting)?),
             "inference-profile-arn" => {
                 if provider != "bedrock" {
                     return Err("inference-profile-arn is only supported for the bedrock provider".to_string());
@@ -57,7 +59,8 @@ impl FromStr for ProviderArg {
             }
             _ => {
                 return Err(
-                    "provider override field must be url, auth, request-model, or inference-profile-arn".to_string()
+                    "provider override field must be url, auth, request-model, idle-timeout-secs, or inference-profile-arn"
+                        .to_string()
                 );
             }
         };
@@ -85,6 +88,10 @@ fn validate_url(url: &str) -> Result<(), String> {
     }
 }
 
+fn parse_idle_timeout_secs(value: &str) -> Result<NonZeroU64, String> {
+    value.parse().map_err(|_| "provider idle timeout must be a positive number of seconds".to_string())
+}
+
 fn parse_auth_mode(value: &str) -> Result<ProviderAuthMode, String> {
     match value {
         "default" => Ok(ProviderAuthMode::Default),
@@ -122,6 +129,12 @@ mod tests {
         assert_eq!(arg.connection.request_model.as_deref(), Some("production-coding"));
     }
 
+    #[test]
+    fn parses_provider_idle_timeout() {
+        let arg: ProviderArg = "ollama.idle-timeout-secs=900".parse().unwrap();
+        assert_eq!(arg.connection.idle_timeout_secs, NonZeroU64::new(900));
+    }
+
     #[test]
     fn combines_repeated_provider_overrides() {
         let args = ProviderConnectionArgs {
@@ -171,6 +184,8 @@ mod tests {
         assert!("bedrock.url=".parse::<ProviderArg>().is_err());
         assert!("bedrock.url=file:///tmp/proxy".parse::<ProviderArg>().is_err());
         assert!("bedrock.auth=disabled".parse::<ProviderArg>().is_err());
+        assert!("ollama.idle-timeout-secs=0".parse::<ProviderArg>().is_err());
+        assert!("ollama.idle-timeout-secs=soon".parse::<ProviderArg>().is_err());
         assert!("bedrock.region=us-west-2".parse::<ProviderArg>().is_err());
         assert!("openai.inference-profile-arn=arn:aws:bedrock:us-west-2:000000000000:application-inference-profile/000000000000".parse::<ProviderArg>().is_err());
     }

crates/llm/Cargo.toml

@@ -81,6 +81,7 @@ aether-doctest = { path = "../doctest" }
 axum = { workspace = true }
 http = { workspace = true }
 tempfile = { workspace = true }
+tokio = { workspace = true, features = ["test-util"] }
 tracing-subscriber = { workspace = true }
 
 [[example]]

crates/llm/src/docs/llm_error.md

@@ -6,9 +6,6 @@ Errors that can occur when interacting with LLM providers.
 - **`MissingApiKey`** -- A required environment variable (e.g. `ANTHROPIC_API_KEY`) is not set.
 - **`OAuthError`** -- OAuth authentication failed (when the `oauth` feature is enabled).
 
-## Request/Response
-- **`HttpClientCreation`** -- Failed to create the HTTP client (e.g. TLS configuration error).
-
 ## Parsing
 - **`JsonParsing`** -- Failed to parse or serialize JSON (response body, tool arguments).
 - **`ToolParameterParsing`** -- Failed to parse tool parameters for a specific tool.

crates/llm/src/docs/provider_connection_override.md

@@ -20,6 +20,12 @@ Request-target routing keeps the catalog model identity and its capabilities whi
 
 Use `url` for providers with resource-specific endpoints, such as Microsoft Foundry. A trailing slash is normalized before `/chat/completions` is appended. A provider with `requestModel` cannot appear more than once in an alloy because the target would be ambiguous.
 
+Give a slow local model more time to produce its first token. A request fails with a retryable timeout once the provider has been silent for `idleTimeoutSecs` (default 300), before responding or between stream events:
+
+```json
+{ "idleTimeoutSecs": 900 }
+```
+
 Pin a Bedrock application inference profile:
 
 ```json

crates/llm/src/error.rs

@@ -12,9 +12,6 @@ pub enum LlmError {
     /// Environment variable not set or invalid
     #[error("{0} environment variable not set")]
     MissingApiKey(String),
-    /// HTTP client creation failed
-    #[error("Failed to create HTTP client: {0}")]
-    HttpClientCreation(String),
     /// Normalized provider-side failure carrying retry classification and
     /// support diagnostics (HTTP status, request ID, provider error code).
     #[error("{0}")]
@@ -328,7 +325,6 @@ mod tests {
         assert!(!LlmError::from(ProviderError::api("x")).is_retryable());
         assert!(!LlmError::from(ProviderError::authentication("x")).is_retryable());
         assert!(!LlmError::MissingApiKey("x".into()).is_retryable());
-        assert!(!LlmError::HttpClientCreation("x".into()).is_retryable());
         assert!(!LlmError::IoError("x".into()).is_retryable());
         assert!(!LlmError::JsonParsing("x".into()).is_retryable());
         assert!(!LlmError::ToolParameterParsing { tool_name: "t".into(), error: "e".into() }.is_retryable());

crates/llm/src/parser.rs

@@ -312,6 +312,7 @@ mod tests {
                     auth_mode: Some(crate::ProviderAuthMode::None),
                     request_model: Some("production-coding".to_string()),
                     inference_profile_arn: None,
+                    idle_timeout_secs: None,
                 },
             )]),
         ));

crates/llm/src/provider.rs

@@ -5,7 +5,7 @@ use crate::Result as LlmResult;
 use crate::catalog::ReasoningEffortError;
 use std::future::Future;
 use std::pin::Pin;
-use tokio_stream::{Stream, StreamExt};
+use tokio_stream::Stream;
 use utils::ReasoningEffort;
 
 use super::{Context, LlmResponse};
@@ -82,37 +82,6 @@ pub(crate) fn validate_reasoning(context: &Context, model: Option<&LlmModel>) ->
     Ok(())
 }
 
-/// Bridge a fallible request setup into an [`LlmResponseStream`].
-///
-/// `open` issues the request; `process` turns what it returns into a response
-/// stream. A setup failure becomes the stream's single item, so providers never
-/// hand-roll the yield-then-return dance and cannot drop an error on the way.
-pub(crate) fn stream_from<T, S>(
-    open: impl Future<Output = LlmResult<T>> + Send + 'static,
-    process: impl FnOnce(T) -> S + Send + 'static,
-) -> LlmResponseStream
-where
-    T: Send,
-    S: Stream<Item = LlmResult<LlmResponse>> + Send + 'static,
-{
-    Box::pin(async_stream::stream! {
-        match open.await {
-            Ok(opened) => {
-                let mut stream = Box::pin(process(opened));
-                while let Some(item) = stream.next().await {
-                    yield item;
-                }
-            }
-            Err(error) => yield Err(error),
-        }
-    })
-}
-
-/// A response stream whose only item is `error`.
-pub(crate) fn error_stream(error: LlmError) -> LlmResponseStream {
-    Box::pin(tokio_stream::once(Err(error)))
-}
-
 impl StreamingModelProvider for Box<dyn StreamingModelProvider> {
     fn stream_response(&self, context: &Context) -> LlmResponseStream {
         (**self).stream_response(context)

crates/llm/src/provider_connection.rs

@@ -1,4 +1,8 @@
 use std::collections::BTreeMap;
+use std::num::NonZeroU64;
+use std::time::Duration;
+
+pub(crate) const DEFAULT_STREAM_IDLE_TIMEOUT: Duration = Duration::from_mins(5);
 
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize, schemars::JsonSchema)]
 #[serde(rename_all = "kebab-case")]
@@ -8,12 +12,13 @@ pub enum ProviderAuthMode {
     None,
 }
 
-#[derive(Clone, Debug, Default, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ProviderConnectionConfig {
     pub base_url: Option<String>,
     pub auth_mode: ProviderAuthMode,
     pub request_model: Option<String>,
     pub inference_profile_arn: Option<String>,
+    pub idle_timeout: Duration,
 }
 
 #[doc = include_str!("docs/provider_connection_override.md")]
@@ -33,6 +38,8 @@ pub struct ProviderConnectionOverride {
     /// AWS Bedrock application inference profile ARN to route requests through.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub inference_profile_arn: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub idle_timeout_secs: Option<NonZeroU64>,
 }
 
 #[derive(Clone, Debug, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize, schemars::JsonSchema)]
@@ -48,6 +55,21 @@ impl ProviderConnectionConfig {
             auth_mode: value.auth_mode.unwrap_or_default(),
             request_model: value.request_model,
             inference_profile_arn: value.inference_profile_arn,
+            idle_timeout: value
+                .idle_timeout_secs
+                .map_or(DEFAULT_STREAM_IDLE_TIMEOUT, |secs| Duration::from_secs(secs.get())),
+        }
+    }
+}
+
+impl Default for ProviderConnectionConfig {
+    fn default() -> Self {
+        Self {
+            base_url: None,
+            auth_mode: ProviderAuthMode::default(),
+            request_model: None,
+            inference_profile_arn: None,
+            idle_timeout: DEFAULT_STREAM_IDLE_TIMEOUT,
         }
     }
 }
@@ -69,6 +91,10 @@ impl ProviderConnectionOverride {
         Self { inference_profile_arn: Some(arn.into()), ..Self::default() }
     }
 
+    pub fn idle_timeout_secs(secs: NonZeroU64) -> Self {
+        Self { idle_timeout_secs: Some(secs), ..Self::default() }
+    }
+
     pub fn merge(&mut self, override_value: Self) {
         if override_value.base_url.is_some() {
             self.base_url = override_value.base_url;
@@ -82,6 +108,9 @@ impl ProviderConnectionOverride {
         if override_value.inference_profile_arn.is_some() {
             self.inference_profile_arn = override_value.inference_profile_arn;
         }
+        if override_value.idle_timeout_secs.is_some() {
+            self.idle_timeout_secs = override_value.idle_timeout_secs;
+        }
     }
 }
 
@@ -142,6 +171,16 @@ mod tests {
         );
     }
 
+    #[test]
+    fn idle_timeout_defaults_and_deserializes() {
+        let overrides: ProviderConnectionOverrides =
+            serde_json::from_str(r#"{"ollama":{"idleTimeoutSecs":900}}"#).unwrap();
+
+        assert_eq!(overrides.config_for("ollama").idle_timeout, Duration::from_mins(15));
+        assert_eq!(overrides.config_for("anthropic").idle_timeout, Duration::from_mins(5));
+        assert!(serde_json::from_str::<ProviderConnectionOverrides>(r#"{"ollama":{"idleTimeoutSecs":0}}"#).is_err());
+    }
+
     #[test]
     fn merge_replaces_inference_profile_arn() {
         let mut first = ProviderConnectionOverride::inference_profile_arn("arn:first");