fix: don't apply request timeout to SSE streams across all SDKs

Long agent runs (>2 min) were being cut mid-stream because each SDK's
HTTP client applied its standard request timeout to SSE response bodies.
The failure surfaces as a decode/connection error, masking the timeout
as the real cause.

- Rust CLI: separate streaming reqwest client with no timeout
- Python: separate httpx client with read=None for post_stream (sync + async)
- TypeScript: dropped AbortController deadline from postStream
- Java: only connectTimeout was set; added Javadoc warnings to lock that in

Default timeout for non-streaming calls bumped 120 -> 900s across all four.

Also adds 'ai-sdk chat --debug [PATH]' to the Rust CLI: writes SSE event
traces (each event, message, and the underlying error + unparsed buffer
when a stream dies) to a file. Defaults to ~/.ai-sdk/chat-debug.log.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: pmbrull

cli/CLAUDE.md

@@ -63,8 +63,16 @@ The `chat` command launches an interactive terminal interface:
 ai-sdk chat              # Opens agent selector
 ai-sdk chat AgentName    # Opens chat with specific agent
 ai-sdk chat AgentName -c <conv-id>  # Resume conversation
+ai-sdk chat --debug                  # Log SSE events to ~/.ai-sdk/chat-debug.log
+ai-sdk chat --debug /tmp/sse.log     # Log SSE events to a custom path
 ```
 
+The `--debug` flag truncates the log on session start and appends events from
+every stream within the session — useful for diagnosing issues like
+`Network error: error decoding response body` mid-stream. The log captures
+each SSE event, message-level details, and the underlying error and unparsed
+buffer when the stream is cut short.
+
 ### TUI Features
 
 - **Agent selection menu**: `/agents` or start without agent name

cli/src/client.rs

@@ -621,9 +621,15 @@ impl MemorySearchResults {
 }
 
 /// AI SDK API client.
+///
+/// Uses two underlying reqwest clients:
+/// - `client`: standard HTTP requests, bounded by `config.timeout`.
+/// - `streaming_client`: SSE endpoints, with no request timeout — agent runs
+///   can take many minutes and the stream itself indicates progress.
 #[derive(Clone)]
 pub struct AISdkClient {
     client: Client,
+    streaming_client: Client,
     base_url: String,
     token: String,
 }
@@ -636,11 +642,18 @@ impl AISdkClient {
             .build()
             .map_err(CliError::from_reqwest)?;
 
+        // No `.timeout(...)` → reqwest waits indefinitely for the response
+        // body, which is what we want for SSE: the server may take minutes
+        // to finish a multi-step agent run, and the stream itself signals
+        // liveness via thinking/message events.
+        let streaming_client = Client::builder().build().map_err(CliError::from_reqwest)?;
+
         // Normalize base URL (remove trailing slash)
         let base_url = config.host.trim_end_matches('/').to_string();
 
         Ok(Self {
             client,
+            streaming_client,
             base_url,
             token: config.token.clone(),
         })
@@ -861,7 +874,7 @@ impl AISdkClient {
         });
 
         let response = self
-            .client
+            .streaming_client
             .post(&url)
             .header("Authorization", self.auth_header())
             .header("Content-Type", "application/json")
@@ -896,7 +909,7 @@ impl AISdkClient {
         };
 
         let response = self
-            .client
+            .streaming_client
             .post(self.agents_url(&format!("/name/{encoded_name}/stream")))
             .header("Authorization", self.auth_header())
             .header("Content-Type", "application/json")

cli/src/commands/chat.rs

@@ -1,7 +1,9 @@
 //! Interactive chat command with TUI interface.
 
+use crate::config;
 use crate::error::CliResult;
 use crate::tui::run_tui;
+use std::path::PathBuf;
 
 /// Run interactive chat session with an agent.
 ///
@@ -10,17 +12,37 @@ use crate::tui::run_tui;
 ///   The TUI calls `stream_default_agent` for each message.
 /// - `agent_name = Some(name)` → use the named dynamic agent (existing behaviour).
 /// - both `None` / `false` → show the agent-selection menu on start.
+///
+/// The `debug` argument carries the raw clap value:
+/// - `None`: debug disabled
+/// - `Some("")`: enabled, write to default path (`~/.ai-sdk/chat-debug.log`)
+/// - `Some(path)`: enabled, write to caller-provided path
 pub async fn run_chat(
     profile: &str,
     agent_name: Option<&str>,
     use_default: bool,
     conversation_id: Option<&str>,
+    debug: Option<String>,
 ) -> CliResult<()> {
+    let debug_log_path: Option<PathBuf> = match debug {
+        None => None,
+        Some(s) if s.is_empty() => Some(config::config_dir()?.join("chat-debug.log")),
+        Some(s) => Some(PathBuf::from(s)),
+    };
+
+    if let Some(path) = debug_log_path.as_ref() {
+        eprintln!(
+            "[ai-sdk] SSE debug events will be written to {}",
+            path.display()
+        );
+    }
+
     run_tui(
         profile,
         agent_name,
         use_default,
         conversation_id.map(String::from),
+        debug_log_path,
     )
     .await
 }

cli/src/config.rs

@@ -44,7 +44,7 @@ impl Default for ProfileConfig {
 }
 
 fn default_timeout() -> u64 {
-    120
+    900
 }
 
 /// Credentials structure (stored separately).

cli/src/main.rs

@@ -131,6 +131,11 @@ enum Commands {
         /// Continue an existing conversation
         #[arg(short = 'c', long = "conversation")]
         conversation_id: Option<String>,
+
+        /// Write SSE debug events to a file. Pass --debug alone to use the
+        /// default path (~/.ai-sdk/chat-debug.log) or --debug PATH to override.
+        #[arg(long, value_name = "PATH", num_args = 0..=1, default_missing_value = "")]
+        debug: Option<String>,
     },
 }
 
@@ -473,12 +478,14 @@ async fn main() {
             agent,
             use_default,
             conversation_id,
+            debug,
         } => {
             commands::chat::run_chat(
                 &cli.profile,
                 agent.as_deref(),
                 use_default,
                 conversation_id.as_deref(),
+                debug,
             )
             .await
         }

cli/src/streaming.rs

@@ -10,6 +10,8 @@ use crate::error::{CliError, CliResult};
 use futures_util::StreamExt;
 use reqwest::Response;
 use serde::Deserialize;
+use std::io::Write;
+use std::path::Path;
 
 /// Sender types matching ChatMessage.
 #[derive(Debug, Clone, PartialEq, Deserialize)]
@@ -167,15 +169,52 @@ pub async fn process_stream<F>(response: Response, handler: F) -> CliResult<Opti
 where
     F: FnMut(ChatMessage),
 {
-    process_stream_with_debug(response, handler, false).await
+    process_stream_inner(response, handler, None).await
 }
 
-/// Process a streaming response with optional debug output.
+/// Process a streaming response with debug events optionally written to stderr.
 pub async fn process_stream_with_debug<F>(
     response: Response,
-    mut handler: F,
+    handler: F,
     debug: bool,
 ) -> CliResult<Option<String>>
+where
+    F: FnMut(ChatMessage),
+{
+    let writer: Option<Box<dyn Write + Send>> = if debug {
+        Some(Box::new(std::io::stderr()))
+    } else {
+        None
+    };
+    process_stream_inner(response, handler, writer).await
+}
+
+/// Process a streaming response and append debug events to a file.
+///
+/// The file is opened in append mode so multiple streams in the same session
+/// accumulate into one log. Callers that want a fresh log per session should
+/// truncate the file once before invoking this.
+pub async fn process_stream_with_debug_file<F>(
+    response: Response,
+    handler: F,
+    debug_log_path: &Path,
+) -> CliResult<Option<String>>
+where
+    F: FnMut(ChatMessage),
+{
+    let file = std::fs::OpenOptions::new()
+        .create(true)
+        .append(true)
+        .open(debug_log_path)?;
+    process_stream_inner(response, handler, Some(Box::new(file))).await
+}
+
+/// Shared implementation for stream processing.
+async fn process_stream_inner<F>(
+    response: Response,
+    mut handler: F,
+    mut debug_writer: Option<Box<dyn Write + Send>>,
+) -> CliResult<Option<String>>
 where
     F: FnMut(ChatMessage),
 {
@@ -186,7 +225,28 @@ where
     let mut message_count = 0;
 
     while let Some(chunk) = stream.next().await {
-        let chunk = chunk.map_err(|e| CliError::NetworkError(e.to_string()))?;
+        let chunk = match chunk {
+            Ok(c) => c,
+            Err(e) => {
+                // Surface where the stream died — the most useful detail when
+                // debugging "error decoding response body" failures.
+                if let Some(w) = debug_writer.as_mut() {
+                    let _ = writeln!(
+                        w,
+                        "[DEBUG] Stream error after {event_count} events, {message_count} messages: {e}"
+                    );
+                    if !buffer.is_empty() {
+                        let _ = writeln!(
+                            w,
+                            "[DEBUG] Unparsed buffer at error ({} bytes): {buffer:?}",
+                            buffer.len()
+                        );
+                    }
+                    let _ = w.flush();
+                }
+                return Err(CliError::NetworkError(e.to_string()));
+            }
+        };
         let text = String::from_utf8_lossy(&chunk);
         buffer.push_str(&text);
 
@@ -197,8 +257,8 @@ where
 
             if let Some((event_type, payload)) = parse_event(&event_str) {
                 event_count += 1;
-                if debug {
-                    eprintln!("[DEBUG] Event #{event_count}: {event_type:?}");
+                if let Some(w) = debug_writer.as_mut() {
+                    let _ = writeln!(w, "[DEBUG] Event #{event_count}: {event_type:?}");
                 }
 
                 // Capture conversation ID from any event
@@ -211,8 +271,9 @@ where
                     if let Some(data) = payload.data {
                         if let Some(message) = data.message {
                             message_count += 1;
-                            if debug {
-                                eprintln!(
+                            if let Some(w) = debug_writer.as_mut() {
+                                let _ = writeln!(
+                                    w,
                                     "[DEBUG] Message #{}: sender={:?}, content_blocks={}",
                                     message_count,
                                     message.sender,
@@ -231,8 +292,8 @@ where
     if !buffer.trim().is_empty() {
         if let Some((event_type, payload)) = parse_event(&buffer) {
             event_count += 1;
-            if debug {
-                eprintln!("[DEBUG] Final Event #{event_count}: {event_type:?}");
+            if let Some(w) = debug_writer.as_mut() {
+                let _ = writeln!(w, "[DEBUG] Final Event #{event_count}: {event_type:?}");
             }
 
             if let Some(id) = &payload.conversation_id {
@@ -242,8 +303,9 @@ where
                 if let Some(data) = payload.data {
                     if let Some(message) = data.message {
                         message_count += 1;
-                        if debug {
-                            eprintln!(
+                        if let Some(w) = debug_writer.as_mut() {
+                            let _ = writeln!(
+                                w,
                                 "[DEBUG] Final Message #{}: sender={:?}",
                                 message_count, message.sender
                             );
@@ -255,8 +317,12 @@ where
         }
     }
 
-    if debug {
-        eprintln!("[DEBUG] Stream complete: {event_count} events, {message_count} messages");
+    if let Some(w) = debug_writer.as_mut() {
+        let _ = writeln!(
+            w,
+            "[DEBUG] Stream complete: {event_count} events, {message_count} messages"
+        );
+        let _ = w.flush();
     }
 
     Ok(conversation_id)
@@ -352,4 +418,44 @@ data: {"conversationId":"conv-1","data":{"message":{"sender":"assistant","conten
         let tools = message.tools_used();
         assert_eq!(tools, vec!["searchMetadata"]);
     }
+
+    #[tokio::test]
+    async fn test_process_stream_with_debug_file_writes_events() {
+        use wiremock::matchers::{method, path};
+        use wiremock::{Mock, MockServer, ResponseTemplate};
+
+        let server = MockServer::start().await;
+        let body = "event: stream-start\ndata: {\"streamId\":\"s1\",\"conversationId\":\"c1\",\"sequence\":0}\n\nevent: message\ndata: {\"conversationId\":\"c1\",\"data\":{\"message\":{\"sender\":\"assistant\",\"content\":[{\"textMessage\":\"hi\"}],\"conversationId\":\"c1\"}}}\n\nevent: stream-completed\ndata: {\"message\":\"done\",\"type\":\"completed\"}\n\n";
+
+        Mock::given(method("GET"))
+            .and(path("/sse"))
+            .respond_with(
+                ResponseTemplate::new(200)
+                    .insert_header("content-type", "text/event-stream")
+                    .set_body_string(body),
+            )
+            .mount(&server)
+            .await;
+
+        let response = reqwest::get(format!("{}/sse", server.uri())).await.unwrap();
+
+        let log_path =
+            std::env::temp_dir().join(format!("ai-sdk-test-debug-{}.log", uuid::Uuid::new_v4()));
+        // Ensure clean slate
+        let _ = std::fs::remove_file(&log_path);
+
+        let conv_id = process_stream_with_debug_file(response, |_msg| {}, &log_path)
+            .await
+            .unwrap();
+        assert_eq!(conv_id.as_deref(), Some("c1"));
+
+        let contents = std::fs::read_to_string(&log_path).unwrap();
+        assert!(contents.contains("Event #1: StreamStart"));
+        assert!(contents.contains("Event #2: Message"));
+        assert!(contents.contains("Message #1: sender=Assistant"));
+        assert!(contents.contains("Event #3: StreamCompleted"));
+        assert!(contents.contains("Stream complete: 3 events, 1 messages"));
+
+        let _ = std::fs::remove_file(&log_path);
+    }
 }

java/src/main/java/io/openmetadata/ai/AISdk.java

@@ -35,7 +35,11 @@
  */
 public class AISdk implements AutoCloseable {
 
-  private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(120);
+  // Generous default — agent runs can take many minutes. Note this maps to
+  // HttpClient.Builder.connectTimeout(...) only (TCP connect), so SSE bodies
+  // are never bounded by it. Per-request HttpRequest.timeout(...) is
+  // intentionally NOT set on streaming requests; see AISdkHttpClient.
+  private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(900);
   private static final int DEFAULT_MAX_RETRIES = 3;
   private static final Duration DEFAULT_RETRY_DELAY = Duration.ofSeconds(1);
   private static final String MEMORIES_BASE_PATH = "/api/v1/contextCenter/memories";
@@ -172,7 +176,13 @@ public Builder token(String token) {
     /**
      * Sets the request timeout.
      *
-     * <p>Default: 120 seconds
+     * <p>This value is applied as the {@link
+     * java.net.http.HttpClient.Builder#connectTimeout(Duration) TCP connect timeout} on the
+     * underlying {@link java.net.http.HttpClient}. It does NOT bound the time spent reading a
+     * response body: SSE streams from long-running agent runs may continue for many minutes, and
+     * the stream's own events (thinking/message/tool) signal liveness.
+     *
+     * <p>Default: 900 seconds
      *
      * @param timeout the timeout duration
      * @return this builder