fix(streaming): prevent SSE idle disconnects and total stream timeouts

config.toml.example

@@ -14,7 +14,7 @@ base_url = "http://127.0.0.1:4010/v1"
 wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
-stream_idle_timeout_ms = 120000
+stream_idle_timeout_ms = 360000
 
 [model_providers.opencode_go_adapter.auth]
 command = "codex-opencode-adapter"

scripts/install-user-provider.ps1

@@ -17,7 +17,7 @@ base_url = "http://127.0.0.1:4010/v1"
 wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
-stream_idle_timeout_ms = 120000
+stream_idle_timeout_ms = 360000
 
 [model_providers.opencode_go_adapter.auth]
 command = "codex-opencode-adapter"

src/config.rs

@@ -8,6 +8,7 @@ pub const DEFAULT_UPSTREAM_BASE: &str = "https://opencode.ai/zen/go/v1";
 pub const DEFAULT_STATE_DB: &str = ".codex-opencode/state.sqlite";
 pub const DEFAULT_STATE_TTL_SECONDS: i64 = 21_600;
 pub const DEFAULT_TIMEOUT_SECONDS: u64 = 300;
+pub const DEFAULT_STREAM_IDLE_TIMEOUT_MS: i64 = 360_000;
 pub const DEFAULT_MAX_REQUEST_BYTES: usize = 8 * 1024 * 1024;
 pub const DEFAULT_MAX_CONCURRENCY: usize = 8;
 

src/init.rs

@@ -1,7 +1,7 @@
 use crate::cli::InitArgs;
 use crate::config::{
     DEFAULT_HOST, DEFAULT_MAX_CONCURRENCY, DEFAULT_MAX_REQUEST_BYTES, DEFAULT_STATE_DB,
-    DEFAULT_STATE_TTL_SECONDS, DEFAULT_TIMEOUT_SECONDS,
+    DEFAULT_STATE_TTL_SECONDS, DEFAULT_STREAM_IDLE_TIMEOUT_MS, DEFAULT_TIMEOUT_SECONDS,
 };
 use crate::project::{generate_project_id, project_key_from_id, ProjectPaths, ProjectRegistry};
 use anyhow::{anyhow, Context};
@@ -166,7 +166,7 @@ fn build_global_codex_config(path: &Path, port: u16) -> anyhow::Result<String> {
     provider["wire_api"] = value("responses");
     provider["request_max_retries"] = value(0);
     provider["stream_max_retries"] = value(0);
-    provider["stream_idle_timeout_ms"] = value(120000);
+    provider["stream_idle_timeout_ms"] = value(DEFAULT_STREAM_IDLE_TIMEOUT_MS);
 
     let auth = ensure_subtable(provider, "auth")?;
     auth["command"] = value("codex-opencode-adapter");
@@ -197,6 +197,28 @@ fn ensure_subtable<'a>(table: &'a mut Table, key: &str) -> anyhow::Result<&'a mu
         .ok_or_else(|| anyhow!("{key} must be a TOML table"))
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn generated_provider_idle_timeout_exceeds_upstream_request_timeout() {
+        let path = std::env::temp_dir().join(format!(
+            "codex-opencode-config-{}.toml",
+            Uuid::new_v4().simple()
+        ));
+        let rendered = build_global_codex_config(&path, crate::config::DEFAULT_PORT).unwrap();
+        let document = rendered.parse::<DocumentMut>().unwrap();
+        let idle_timeout = document["model_providers"]["opencode_go_adapter"]
+            ["stream_idle_timeout_ms"]
+            .as_integer()
+            .unwrap();
+
+        assert_eq!(idle_timeout, DEFAULT_STREAM_IDLE_TIMEOUT_MS);
+        assert!(idle_timeout > (DEFAULT_TIMEOUT_SECONDS * 1_000) as i64);
+    }
+}
+
 fn create_backup_if_exists(path: &Path) -> anyhow::Result<Option<PathBuf>> {
     if !path.exists() {
         return Ok(None);

src/server.rs

@@ -1,5 +1,6 @@
 use axum::extract::State;
 use axum::http::{HeaderMap, StatusCode};
+use axum::response::sse::KeepAlive;
 use axum::response::{IntoResponse, Response, Sse};
 use axum::routing::{get, post};
 use axum::{Json, Router};
@@ -9,6 +10,7 @@ use std::collections::HashMap;
 use std::convert::Infallible;
 use std::path::PathBuf;
 use std::sync::{Arc, RwLock};
+use std::time::Duration;
 use tokio::sync::Semaphore;
 use uuid::Uuid;
 
@@ -31,6 +33,12 @@ use crate::upstream::{
     OpenCodeGoClient, UpstreamError,
 };
 
+const DOWNSTREAM_SSE_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(15);
+
+fn downstream_sse_keep_alive(interval: Duration) -> KeepAlive {
+    KeepAlive::new().interval(interval).text("keep-alive")
+}
+
 #[derive(Clone)]
 pub struct ProjectRuntime {
     pub config: Config,
@@ -541,7 +549,11 @@ async fn stream_response(
     });
 
     let stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
-    Sse::new(stream).into_response()
+    Sse::new(stream)
+        .keep_alive(downstream_sse_keep_alive(
+            DOWNSTREAM_SSE_KEEP_ALIVE_INTERVAL,
+        ))
+        .into_response()
 }
 
 fn previous_response(
@@ -932,3 +944,27 @@ fn upstream_error(error: UpstreamError) -> Response {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn downstream_sse_emits_keep_alive_while_response_stream_is_idle() {
+        let stream = futures::stream::pending::<
+            Result<axum::response::sse::Event, std::convert::Infallible>,
+        >();
+        let response = Sse::new(stream)
+            .keep_alive(downstream_sse_keep_alive(Duration::from_millis(5)))
+            .into_response();
+        let mut body = response.into_body().into_data_stream();
+
+        let chunk = tokio::time::timeout(Duration::from_millis(100), body.next())
+            .await
+            .expect("keep-alive should arrive before the test timeout")
+            .expect("SSE body should still be open")
+            .expect("keep-alive body chunk should be readable");
+
+        assert_eq!(std::str::from_utf8(&chunk).unwrap(), ": keep-alive\n\n");
+    }
+}

src/upstream.rs

@@ -21,6 +21,7 @@ pub struct OpenCodeGoClient {
     base_url: String,
     api_key: String,
     client: reqwest::Client,
+    stream_client: reqwest::Client,
 }
 
 impl OpenCodeGoClient {
@@ -29,11 +30,14 @@ impl OpenCodeGoClient {
         api_key: impl Into<String>,
         timeout_seconds: u64,
     ) -> anyhow::Result<Self> {
+        let timeout = Duration::from_secs(timeout_seconds);
         Ok(Self {
             base_url: base_url.into().trim_end_matches('/').to_string(),
             api_key: api_key.into(),
-            client: reqwest::Client::builder()
-                .timeout(Duration::from_secs(timeout_seconds))
+            client: reqwest::Client::builder().timeout(timeout).build()?,
+            stream_client: reqwest::Client::builder()
+                .connect_timeout(timeout)
+                .read_timeout(timeout)
                 .build()?,
         })
     }
@@ -55,7 +59,7 @@ impl OpenCodeGoClient {
     ) -> Result<impl Stream<Item = Result<Bytes, reqwest::Error>>, UpstreamError> {
         payload["stream"] = Value::Bool(true);
         let response = self
-            .client
+            .stream_client
             .post(format!("{}/chat/completions", self.base_url))
             .headers(self.headers("text/event-stream")?)
             .json(&payload)
@@ -213,6 +217,7 @@ pub fn sse_data_from_block(block: &str) -> Option<String> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use futures::StreamExt as _;
     use serde_json::json;
 
     // ── extract_error_message ──
@@ -383,4 +388,42 @@ mod tests {
         let data = sse_data_from_block(block).unwrap();
         assert_eq!(data, "{\"key\":\"value\"}");
     }
+
+    #[tokio::test]
+    async fn streaming_request_can_outlive_nonstream_total_timeout() {
+        async fn slow_stream() -> axum::response::Sse<
+            impl futures::Stream<Item = Result<axum::response::sse::Event, std::convert::Infallible>>,
+        > {
+            let stream = async_stream::stream! {
+                for index in 0..4 {
+                    tokio::time::sleep(Duration::from_millis(350)).await;
+                    yield Ok(axum::response::sse::Event::default()
+                        .data(json!({"chunk": index}).to_string()));
+                }
+            };
+            axum::response::Sse::new(stream)
+        }
+
+        let app = axum::Router::new().route("/chat/completions", axum::routing::post(slow_stream));
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+
+        let client = OpenCodeGoClient::new(format!("http://{address}"), "test-key", 1).unwrap();
+        let started = std::time::Instant::now();
+        let mut stream = client
+            .chat_stream(json!({"model": "test-model"}))
+            .await
+            .unwrap();
+        let mut chunks = 0;
+        while let Some(chunk) = stream.next().await {
+            chunk.expect("each SSE body chunk should arrive before the read timeout");
+            chunks += 1;
+        }
+
+        assert!(started.elapsed() > Duration::from_secs(1));
+        assert!(chunks >= 4);
+    }
 }