Surface stream failures instead of truncating the answer

crates/api/src/lib.rs

@@ -440,14 +440,27 @@ pub async fn init_domain_services_with_pool(
         as Arc<dyn services::completions::ports::OrganizationConcurrentLimitRepository>;
 
     // Create completion service with usage tracking (needs usage_service)
-    let completion_service = Arc::new(services::CompletionServiceImpl::new(
+    let mut completion_service_impl = services::CompletionServiceImpl::new(
         inference_provider_pool.clone(),
         attestation_service.clone(),
         usage_service.clone(),
         metrics_service.clone(),
         models_repo.clone() as Arc<dyn services::models::ModelsRepository>,
         org_limit_repository,
-    ));
+    );
+    if config.stream_watchdog.enabled {
+        completion_service_impl = completion_service_impl.with_stream_idle_timeouts(
+            services::completions::StreamIdleTimeouts {
+                first_token: std::time::Duration::from_secs(
+                    config.stream_watchdog.first_token_seconds,
+                ),
+                between_tokens: std::time::Duration::from_secs(
+                    config.stream_watchdog.between_tokens_seconds,
+                ),
+            },
+        );
+    }
+    let completion_service = Arc::new(completion_service_impl);
 
     let brave_search_provider =
         Arc::new(services::responses::tools::brave::BraveWebSearchProvider::new());
@@ -2824,6 +2837,7 @@ mod tests {
             staking_farm: config::StakingFarmConfig::default(),
             aml: config::AmlConfig::default(),
             usage_reporting: config::UsageReportingConfig::default(),
+            stream_watchdog: config::StreamWatchdogConfig::default(),
             ita: config::ItaAttestationConfig::default(),
         };
 
@@ -2938,6 +2952,7 @@ mod tests {
             staking_farm: config::StakingFarmConfig::default(),
             aml: config::AmlConfig::default(),
             usage_reporting: config::UsageReportingConfig::default(),
+            stream_watchdog: config::StreamWatchdogConfig::default(),
             ita: config::ItaAttestationConfig::default(),
         };
 

crates/api/src/routes/completions.rs

@@ -320,6 +320,10 @@ fn completion_stream_error_category(e: &inference_providers::CompletionError) ->
     }
 }
 
+fn sanitized_stream_error(e: &inference_providers::CompletionError) -> String {
+    services::inference_provider_pool::InferenceProviderPool::safe_error_detail(e)
+}
+
 /// Returns an OpenAI-compatible `error.type` for a stream-level completion error.
 /// Used in the `data: {"error":{...}}` SSE frame so clients can branch on the type.
 ///
@@ -1848,10 +1852,13 @@ async fn chat_completions_inner(
                                     let count = error_count_inner
                                         .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                                     if count == 0 {
+                                        let error_detail = sanitized_stream_error(&e);
                                         tracing::error!(
+                                            %request_id,
                                             %organization_id,
                                             model = %model_for_err,
                                             error_type = %completion_stream_error_category(&e),
+                                            %error_detail,
                                             "Completion stream error"
                                         );
                                     }
@@ -2492,10 +2499,13 @@ async fn completions_inner(
                                 )))
                             }),
                             Err(e) => {
+                                let error_detail = sanitized_stream_error(&e);
                                 tracing::error!(
+                                    %request_id,
                                     %organization_id,
                                     model = %model_for_err,
                                     error_type = %completion_stream_error_category(&e),
+                                    %error_detail,
                                     "Text completion stream error"
                                 );
                                 Some(Ok::<Bytes, Infallible>(sse_error_frame(&e)))

crates/api/tests/common/mod.rs

@@ -136,6 +136,7 @@ pub fn test_config() -> ApiConfig {
             enabled: true,
             ..config::UsageReportingConfig::default()
         },
+        stream_watchdog: config::StreamWatchdogConfig::default(),
         ita: config::ItaAttestationConfig::default(),
     }
 }

crates/config/src/types.rs

@@ -35,6 +35,7 @@ pub struct ApiConfig {
     pub staking_farm: StakingFarmConfig,
     pub aml: AmlConfig,
     pub usage_reporting: UsageReportingConfig,
+    pub stream_watchdog: StreamWatchdogConfig,
     pub ita: ItaAttestationConfig,
 }
 
@@ -79,6 +80,7 @@ impl ApiConfig {
             aml: AmlConfig::from_env()?,
             ita: ItaAttestationConfig::from_env()?,
             usage_reporting: UsageReportingConfig::from_env()?,
+            stream_watchdog: StreamWatchdogConfig::from_env()?,
         })
     }
 }
@@ -304,6 +306,57 @@ fn parse_optional_i32_env(key: &str, default: Option<i32>) -> Result<Option<i32>
     }
 }
 
+/// Idle bounds that fail a stream producing no data, off unless an operator
+/// enables it. The first-token bound is separate because a large context can
+/// prefill for minutes before any token appears.
+#[derive(Debug, Clone)]
+pub struct StreamWatchdogConfig {
+    pub enabled: bool,
+    pub first_token_seconds: u64,
+    pub between_tokens_seconds: u64,
+}
+
+impl Default for StreamWatchdogConfig {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            first_token_seconds: 300,
+            between_tokens_seconds: 90,
+        }
+    }
+}
+
+impl StreamWatchdogConfig {
+    pub fn from_env() -> Result<Self, String> {
+        let defaults = Self::default();
+        let config = Self {
+            enabled: parse_bool_env("STREAM_WATCHDOG_ENABLED", defaults.enabled)?,
+            first_token_seconds: parse_u64_env(
+                "STREAM_WATCHDOG_FIRST_TOKEN_SECONDS",
+                defaults.first_token_seconds,
+            )?,
+            between_tokens_seconds: parse_u64_env(
+                "STREAM_WATCHDOG_BETWEEN_TOKENS_SECONDS",
+                defaults.between_tokens_seconds,
+            )?,
+        };
+
+        if config.first_token_seconds == 0 || config.between_tokens_seconds == 0 {
+            return Err("stream watchdog timeouts must be greater than zero".to_string());
+        }
+        if config.first_token_seconds > 3_600 || config.between_tokens_seconds > 3_600 {
+            return Err("stream watchdog timeouts must not exceed 3600".to_string());
+        }
+        if config.first_token_seconds < config.between_tokens_seconds {
+            return Err("STREAM_WATCHDOG_FIRST_TOKEN_SECONDS must not be below \
+                 STREAM_WATCHDOG_BETWEEN_TOKENS_SECONDS"
+                .to_string());
+        }
+
+        Ok(config)
+    }
+}
+
 /// Operational limits for the programmatic usage-reporting API.
 ///
 /// Reporting is disabled by default because its production indexes are built
@@ -1255,6 +1308,87 @@ mod tests {
         );
     }
 
+    struct StreamWatchdogEnvGuard {
+        values: [(&'static str, Option<OsString>); 3],
+    }
+
+    impl StreamWatchdogEnvGuard {
+        const KEYS: [&'static str; 3] = [
+            "STREAM_WATCHDOG_ENABLED",
+            "STREAM_WATCHDOG_FIRST_TOKEN_SECONDS",
+            "STREAM_WATCHDOG_BETWEEN_TOKENS_SECONDS",
+        ];
+
+        fn cleared() -> Self {
+            let guard = Self {
+                values: Self::KEYS.map(|key| (key, std::env::var_os(key))),
+            };
+            for key in Self::KEYS {
+                std::env::remove_var(key);
+            }
+            guard
+        }
+    }
+
+    impl Drop for StreamWatchdogEnvGuard {
+        fn drop(&mut self) {
+            for (key, value) in &mut self.values {
+                match value.take() {
+                    Some(value) => std::env::set_var(*key, value),
+                    None => std::env::remove_var(*key),
+                }
+            }
+        }
+    }
+
+    #[test]
+    #[serial]
+    fn stream_watchdog_is_off_by_default_with_a_longer_first_token_bound() {
+        let _env = StreamWatchdogEnvGuard::cleared();
+
+        let config = StreamWatchdogConfig::from_env().unwrap();
+
+        assert!(!config.enabled);
+        assert_eq!(config.first_token_seconds, 300);
+        assert_eq!(config.between_tokens_seconds, 90);
+        assert!(config.first_token_seconds >= config.between_tokens_seconds);
+    }
+
+    #[test]
+    #[serial]
+    fn stream_watchdog_rejects_a_first_token_bound_below_the_between_token_bound() {
+        let _env = StreamWatchdogEnvGuard::cleared();
+        std::env::set_var("STREAM_WATCHDOG_FIRST_TOKEN_SECONDS", "30");
+        std::env::set_var("STREAM_WATCHDOG_BETWEEN_TOKENS_SECONDS", "90");
+
+        let error = StreamWatchdogConfig::from_env().unwrap_err();
+
+        assert!(error.contains("STREAM_WATCHDOG_FIRST_TOKEN_SECONDS"));
+        assert!(error.contains("STREAM_WATCHDOG_BETWEEN_TOKENS_SECONDS"));
+    }
+
+    #[test]
+    #[serial]
+    fn stream_watchdog_rejects_a_zero_bound_that_would_fail_every_stream() {
+        let _env = StreamWatchdogEnvGuard::cleared();
+        std::env::set_var("STREAM_WATCHDOG_BETWEEN_TOKENS_SECONDS", "0");
+
+        let error = StreamWatchdogConfig::from_env().unwrap_err();
+
+        assert!(error.contains("greater than zero"));
+    }
+
+    #[test]
+    #[serial]
+    fn stream_watchdog_rejects_a_bound_beyond_an_hour() {
+        let _env = StreamWatchdogEnvGuard::cleared();
+        std::env::set_var("STREAM_WATCHDOG_FIRST_TOKEN_SECONDS", "3601");
+
+        let error = StreamWatchdogConfig::from_env().unwrap_err();
+
+        assert!(error.contains("3600"));
+    }
+
     #[test]
     fn test_is_admin_email() {
         let config = AuthConfig {

crates/inference_providers/src/attested/nearai/mod.rs

@@ -95,7 +95,7 @@ fn format_error_chain<E: std::error::Error>(e: &E) -> String {
 /// strips; the corresponding HTTP header names are `X-Request-Id`, `X-Org-Id`,
 /// and `X-Workspace-Id`. Exposed as `pub(crate)` so `external/mod.rs` can use
 /// the same constants instead of hardcoding the strings.
-pub(crate) mod tracing_headers {
+pub mod tracing_headers {
     /// UUIDv4 generated per request by cloud-api. Join key across all hops.
     pub const REQUEST_ID: &str = "x_request_id";
     /// Organization UUID of the authenticated API key owner.

crates/services/Cargo.toml

@@ -72,3 +72,4 @@ tokio-test = "0.4"
 async-trait = "0.1"
 futures = "0.3"
 mockall = "0.14"
+tracing-subscriber = { version = "0.3", features = ["json"] }