fix(ai): align stream TTFT with first effective output

src/crates/adapters/ai-adapters/src/client.rs

@@ -37,22 +37,20 @@ pub struct StreamResponse {
     pub trace_handle: Option<ModelExchangeRequestTraceHandle>,
 }
 
-/// Default time to wait for the first response headers / stream body to start.
+/// Default time to wait for the first effective streamed output after a request starts.
 pub const DEFAULT_STREAM_TTFT_TIMEOUT_SECS: u64 = 30;
 
 /// Default idle time between streamed chunks once the stream has started.
 pub const DEFAULT_STREAM_IDLE_TIMEOUT_SECS: u64 = 45;
 
-/// Minimum TTFT for models with explicit reasoning enabled.
-pub const REASONING_STREAM_TTFT_TIMEOUT_SECS: u64 = 45;
-
 /// Runtime stream behavior shared across provider implementations.
 #[derive(Debug, Clone, Default)]
 pub struct StreamOptions {
     /// Maximum idle time between streamed chunks. `None` means wait indefinitely.
     pub idle_timeout: Option<Duration>,
-    /// Maximum time to wait for HTTP response headers when opening a stream.
-    /// `None` means wait indefinitely.
+    /// Maximum time to wait for the first effective streamed output (text,
+    /// reasoning, or tool-call data) after a request starts. `None` means wait
+    /// indefinitely.
     pub ttft_timeout: Option<Duration>,
 }
 
@@ -100,7 +98,7 @@ impl AIClient {
         self.stream_options.idle_timeout
     }
 
-    /// Returns the configured time-to-first-token timeout for opening a stream, if any.
+    /// Returns the configured timeout for the first effective streamed output, if any.
     pub fn stream_ttft_timeout(&self) -> Option<Duration> {
         self.stream_options.ttft_timeout
     }

src/crates/adapters/ai-adapters/src/client/sse.rs

@@ -55,11 +55,18 @@ where
 fn format_ttft_timeout_error(label: &str, ttft_timeout: Option<Duration>) -> String {
     let timeout_secs = ttft_timeout.map(|timeout| timeout.as_secs()).unwrap_or(0);
     format!(
-        "{} TTFT timeout after {}s waiting for response headers",
+        "{} TTFT timeout after {}s waiting for first effective stream output",
         label, timeout_secs
     )
 }
 
+fn remaining_ttft_timeout(
+    started_at: std::time::Instant,
+    ttft_timeout: Option<Duration>,
+) -> Option<Duration> {
+    ttft_timeout.map(|timeout| timeout.saturating_sub(started_at.elapsed()))
+}
+
 fn format_transport_error(label: &str, error: &reqwest::Error) -> String {
     format!("{} connection failed: {}", label, error)
 }
@@ -116,6 +123,7 @@ where
         reqwest::Response,
         mpsc::UnboundedSender<Result<UnifiedResponse>>,
         Option<mpsc::UnboundedSender<String>>,
+        Option<Duration>,
     ),
 {
     let mut last_error = None;
@@ -279,7 +287,8 @@ where
 
         let (tx, rx) = mpsc::unbounded_channel();
         let (tx_raw, rx_raw) = mpsc::unbounded_channel();
-        spawn_handler(response, tx, Some(tx_raw));
+        let remaining_ttft_timeout = remaining_ttft_timeout(request_start_time, ttft_timeout);
+        spawn_handler(response, tx, Some(tx_raw), remaining_ttft_timeout);
 
         return Ok(StreamResponse {
             stream: Box::pin(tokio_stream::wrappers::UnboundedReceiverStream::new(rx)),
@@ -311,6 +320,17 @@ mod tests {
         );
 
         assert!(message.contains("TTFT timeout after 30s"));
+        assert!(message.contains("first effective stream output"));
+    }
+
+    #[test]
+    fn remaining_ttft_timeout_subtracts_elapsed_request_time() {
+        let start = std::time::Instant::now() - Duration::from_secs(2);
+        let remaining = remaining_ttft_timeout(start, Some(Duration::from_secs(5)));
+
+        let remaining = remaining.expect("remaining timeout");
+        assert!(remaining <= Duration::from_secs(3));
+        assert!(remaining > Duration::from_secs(2));
     }
 
     #[test]

src/crates/adapters/ai-adapters/src/lib.rs

@@ -11,7 +11,7 @@ pub mod types;
 
 pub use client::{
     AIClient, StreamOptions, StreamResponse, DEFAULT_STREAM_IDLE_TIMEOUT_SECS,
-    DEFAULT_STREAM_TTFT_TIMEOUT_SECS, REASONING_STREAM_TTFT_TIMEOUT_SECS,
+    DEFAULT_STREAM_TTFT_TIMEOUT_SECS,
 };
 pub use model_selector::{
     classify_model_selector, resolve_cache_model_selector, resolve_required_model_selector,

src/crates/adapters/ai-adapters/src/providers/anthropic/request.rs

@@ -350,12 +350,13 @@ pub(crate) async fn send_stream(
         ttft_timeout,
         trace,
         || apply_headers(client, client.client.post(&url), &url),
-        move |response, tx, tx_raw| {
+        move |response, tx, tx_raw, remaining_ttft_timeout| {
             tokio::spawn(handle_anthropic_stream(
                 response,
                 tx,
                 tx_raw,
                 inline_think_in_text,
+                remaining_ttft_timeout,
                 idle_timeout,
             ));
         },

src/crates/adapters/ai-adapters/src/providers/gemini/code_assist.rs

@@ -182,8 +182,14 @@ pub(crate) async fn send_stream(
         ttft_timeout,
         trace,
         || apply_headers(client, client.client.post(&url)),
-        move |response, tx, tx_raw| {
-            tokio::spawn(handle_gemini_stream(response, tx, tx_raw, idle_timeout));
+        move |response, tx, tx_raw, remaining_ttft_timeout| {
+            tokio::spawn(handle_gemini_stream(
+                response,
+                tx,
+                tx_raw,
+                remaining_ttft_timeout,
+                idle_timeout,
+            ));
         },
     )
     .await

src/crates/adapters/ai-adapters/src/providers/gemini/request.rs

@@ -346,8 +346,14 @@ pub(crate) async fn send_stream(
         ttft_timeout,
         trace,
         || apply_headers(client, client.client.post(&url)),
-        move |response, tx, tx_raw| {
-            tokio::spawn(handle_gemini_stream(response, tx, tx_raw, idle_timeout));
+        move |response, tx, tx_raw, remaining_ttft_timeout| {
+            tokio::spawn(handle_gemini_stream(
+                response,
+                tx,
+                tx_raw,
+                remaining_ttft_timeout,
+                idle_timeout,
+            ));
         },
     )
     .await

src/crates/adapters/ai-adapters/src/providers/openai/chat.rs

@@ -100,12 +100,13 @@ pub(crate) async fn send_stream(
         ttft_timeout,
         trace,
         || common::apply_headers(client, client.client.post(&url)),
-        move |response, tx, tx_raw| {
+        move |response, tx, tx_raw, remaining_ttft_timeout| {
             tokio::spawn(handle_openai_stream(
                 response,
                 tx,
                 tx_raw,
                 inline_think_in_text,
+                remaining_ttft_timeout,
                 idle_timeout,
             ));
         },

src/crates/adapters/ai-adapters/src/providers/openai/codex_chatgpt.rs

@@ -170,8 +170,14 @@ pub(crate) async fn send_stream(
         ttft_timeout,
         trace,
         || common::apply_headers(client, client.client.post(&url)),
-        move |response, tx, tx_raw| {
-            tokio::spawn(handle_responses_stream(response, tx, tx_raw, idle_timeout));
+        move |response, tx, tx_raw, remaining_ttft_timeout| {
+            tokio::spawn(handle_responses_stream(
+                response,
+                tx,
+                tx_raw,
+                remaining_ttft_timeout,
+                idle_timeout,
+            ));
         },
     )
     .await

src/crates/adapters/ai-adapters/src/providers/openai/responses.rs

@@ -134,8 +134,14 @@ pub(crate) async fn send_stream(
         ttft_timeout,
         trace,
         || common::apply_headers(client, client.client.post(&url)),
-        move |response, tx, tx_raw| {
-            tokio::spawn(handle_responses_stream(response, tx, tx_raw, idle_timeout));
+        move |response, tx, tx_raw, remaining_ttft_timeout| {
+            tokio::spawn(handle_responses_stream(
+                response,
+                tx,
+                tx_raw,
+                remaining_ttft_timeout,
+                idle_timeout,
+            ));
         },
     )
     .await

src/crates/adapters/ai-adapters/src/stream/stream_handler/anthropic.rs

@@ -1,6 +1,6 @@
 use super::inline_think::InlineThinkParser;
 use super::stream_stats::StreamStats;
-use super::{next_stream_item, TimedStreamItem};
+use super::{next_stream_item, StreamTimeoutController, StreamTimeoutStage, TimedStreamItem};
 use crate::stream::types::anthropic::{
     AnthropicSSEError, ContentBlock, ContentBlockDelta, ContentBlockStart, MessageDelta,
     MessageStart, Usage,
@@ -26,16 +26,18 @@ pub async fn handle_anthropic_stream(
     tx_event: mpsc::UnboundedSender<Result<UnifiedResponse>>,
     tx_raw_sse: Option<mpsc::UnboundedSender<String>>,
     inline_think_in_text: bool,
+    ttft_timeout: Option<Duration>,
     idle_timeout: Option<Duration>,
 ) {
     let mut stream = response.bytes_stream().eventsource();
     let mut usage = Usage::default();
     let mut stats = StreamStats::new("Anthropic");
     let mut inline_think_parser = InlineThinkParser::new(inline_think_in_text);
+    let mut timeout_controller = StreamTimeoutController::new(ttft_timeout, idle_timeout);
     let mut received_finish_reason = false;
 
     loop {
-        let sse = match next_stream_item(&mut stream, idle_timeout).await {
+        let sse = match next_stream_item(&mut stream, &timeout_controller).await {
             TimedStreamItem::Item(Ok(sse)) => sse,
             TimedStreamItem::End => {
                 if received_finish_reason {
@@ -60,7 +62,18 @@ pub async fn handle_anthropic_stream(
                 let _ = tx_event.send(Err(anyhow!(error_msg)));
                 return;
             }
-            TimedStreamItem::TimedOut => {
+            TimedStreamItem::TimedOut(StreamTimeoutStage::Ttft) => {
+                let timeout_secs = ttft_timeout.map(|timeout| timeout.as_secs()).unwrap_or(0);
+                let error_msg = format!(
+                    "Anthropic stream TTFT timeout after {}s waiting for first effective output",
+                    timeout_secs
+                );
+                stats.log_summary("ttft_timeout");
+                error!("{}", error_msg);
+                let _ = tx_event.send(Err(anyhow!(error_msg)));
+                return;
+            }
+            TimedStreamItem::TimedOut(StreamTimeoutStage::Idle) => {
                 let timeout_secs = idle_timeout.map(|timeout| timeout.as_secs()).unwrap_or(0);
                 let error_msg = format!(
                     "SSE Timeout: idle timeout waiting for SSE after {}s",
@@ -133,6 +146,7 @@ pub async fn handle_anthropic_stream(
                 ) {
                     emit_normalized_response(
                         &mut inline_think_parser,
+                        &mut timeout_controller,
                         &tx_event,
                         &mut stats,
                         UnifiedResponse::from(content_block_start),
@@ -154,6 +168,7 @@ pub async fn handle_anthropic_stream(
                 match UnifiedResponse::try_from(content_block_delta) {
                     Ok(unified_response) => emit_normalized_response(
                         &mut inline_think_parser,
+                        &mut timeout_controller,
                         &tx_event,
                         &mut stats,
                         unified_response,
@@ -188,6 +203,7 @@ pub async fn handle_anthropic_stream(
                 }
                 emit_normalized_response(
                     &mut inline_think_parser,
+                    &mut timeout_controller,
                     &tx_event,
                     &mut stats,
                     unified_response,
@@ -344,11 +360,13 @@ fn trace_unified_response_if_useful(response: &UnifiedResponse) {
 
 fn emit_normalized_response(
     inline_think_parser: &mut InlineThinkParser,
+    timeout_controller: &mut StreamTimeoutController,
     tx_event: &mpsc::UnboundedSender<Result<UnifiedResponse>>,
     stats: &mut StreamStats,
     unified_response: UnifiedResponse,
 ) {
     for normalized_response in inline_think_parser.normalize_response(unified_response) {
+        timeout_controller.observe_unified_response(&normalized_response);
         trace_unified_response_if_useful(&normalized_response);
         stats.record_unified_response(&normalized_response);
         let _ = tx_event.send(Ok(normalized_response));

src/crates/adapters/ai-adapters/src/stream/stream_handler/gemini.rs

@@ -1,5 +1,5 @@
 use super::stream_stats::StreamStats;
-use super::{next_stream_item, TimedStreamItem};
+use super::{next_stream_item, StreamTimeoutController, StreamTimeoutStage, TimedStreamItem};
 use crate::stream::types::gemini::GeminiSSEData;
 use crate::stream::types::unified::UnifiedResponse;
 use anyhow::{anyhow, Result};
@@ -80,15 +80,17 @@ pub async fn handle_gemini_stream(
     response: Response,
     tx_event: mpsc::UnboundedSender<Result<UnifiedResponse>>,
     tx_raw_sse: Option<mpsc::UnboundedSender<String>>,
+    ttft_timeout: Option<Duration>,
     idle_timeout: Option<Duration>,
 ) {
     let mut stream = response.bytes_stream().eventsource();
     let mut received_finish_reason = false;
     let mut tool_call_state = GeminiToolCallState::new();
     let mut stats = StreamStats::new("Gemini");
+    let mut timeout_controller = StreamTimeoutController::new(ttft_timeout, idle_timeout);
 
     loop {
-        let sse = match next_stream_item(&mut stream, idle_timeout).await {
+        let sse = match next_stream_item(&mut stream, &timeout_controller).await {
             TimedStreamItem::Item(Ok(sse)) => sse,
             TimedStreamItem::End => {
                 if received_finish_reason {
@@ -108,7 +110,18 @@ pub async fn handle_gemini_stream(
                 let _ = tx_event.send(Err(anyhow!(error_msg)));
                 return;
             }
-            TimedStreamItem::TimedOut => {
+            TimedStreamItem::TimedOut(StreamTimeoutStage::Ttft) => {
+                let timeout_secs = ttft_timeout.map(|timeout| timeout.as_secs()).unwrap_or(0);
+                let error_msg = format!(
+                    "Gemini stream TTFT timeout after {}s waiting for first effective output",
+                    timeout_secs
+                );
+                stats.log_summary("ttft_timeout");
+                error!("{}", error_msg);
+                let _ = tx_event.send(Err(anyhow!(error_msg)));
+                return;
+            }
+            TimedStreamItem::TimedOut(StreamTimeoutStage::Idle) => {
                 let timeout_secs = idle_timeout.map(|timeout| timeout.as_secs()).unwrap_or(0);
                 let error_msg = format!("Gemini SSE stream timeout after {}s", timeout_secs);
                 stats.log_summary("sse_stream_timeout");
@@ -189,6 +202,7 @@ pub async fn handle_gemini_stream(
         );
 
         for unified_response in unified_responses {
+            timeout_controller.observe_unified_response(&unified_response);
             stats.record_unified_response(&unified_response);
             let _ = tx_event.send(Ok(unified_response));
         }