fix(inference): idle-timeout recovery + guaranteed terminal event on a stalled stream

src/openhuman/inference/provider/compatible.rs

@@ -19,7 +19,7 @@ mod compatible_stream;
 #[path = "compatible_stream_native.rs"]
 mod compatible_stream_native;
 #[path = "compatible_timeout.rs"]
-mod compatible_timeout;
+pub(crate) mod compatible_timeout;
 #[path = "compatible_types.rs"]
 mod compatible_types;
 

src/openhuman/inference/provider/compatible_stream.rs

@@ -4,10 +4,44 @@
 //! `StreamChunk` stream via Server-Sent Events parsing.
 
 use crate::openhuman::inference::provider::traits::{StreamChunk, StreamError, StreamResult};
-use futures_util::{stream, StreamExt};
+use futures_util::{stream, Stream, StreamExt};
+use std::time::Duration;
 
 use super::compatible_parse::parse_sse_line;
 
+/// Outcome of a single idle-bounded read from an item stream.
+#[derive(Debug, PartialEq, Eq)]
+enum IdleRead<T> {
+    /// The stream yielded an item.
+    Item(T),
+    /// The stream ended cleanly (`next()` returned `None`).
+    Ended,
+    /// No item arrived within the idle timeout — the upstream stalled.
+    IdleTimedOut,
+}
+
+/// Read the next item from `stream`, enforcing an optional idle timeout.
+///
+/// Extracted as a small generic helper (over the concrete
+/// `reqwest::Response::bytes_stream()`) so the stall→terminal-error decision is
+/// unit-testable without standing up a live HTTP response.
+async fn read_next_or_idle<S, T>(stream: &mut S, idle: Option<Duration>) -> IdleRead<T>
+where
+    S: Stream<Item = T> + Unpin,
+{
+    match idle {
+        Some(dur) => match tokio::time::timeout(dur, stream.next()).await {
+            Ok(Some(item)) => IdleRead::Item(item),
+            Ok(None) => IdleRead::Ended,
+            Err(_elapsed) => IdleRead::IdleTimedOut,
+        },
+        None => match stream.next().await {
+            Some(item) => IdleRead::Item(item),
+            None => IdleRead::Ended,
+        },
+    }
+}
+
 /// Convert SSE byte stream to text chunks.
 pub(crate) fn sse_bytes_to_chunks(
     response: reqwest::Response,
@@ -30,8 +64,36 @@ pub(crate) fn sse_bytes_to_chunks(
         }
 
         let mut bytes_stream = response.bytes_stream();
+        // Reuse the single shared per-chunk idle watchdog (#4269) rather than a
+        // second, divergent resolver: one bounded default (1..=3600s, documented
+        // in `.env.example`) drives both this SSE path and the native stream
+        // parser, so `OPENHUMAN_INFERENCE_STREAM_IDLE_TIMEOUT_SECS` means the same
+        // thing everywhere and a typo can never wedge a turn indefinitely.
+        // Absolute path: `compatible_stream` is included as a module under both
+        // `provider` and `provider::compatible`, so a `super::`-relative path
+        // resolves differently in each and can't work in both.
+        let idle_timeout = Some(
+            crate::openhuman::inference::provider::compatible::compatible_timeout::stream_idle_timeout(),
+        );
 
-        while let Some(item) = bytes_stream.next().await {
+        loop {
+            let item = match read_next_or_idle(&mut bytes_stream, idle_timeout).await {
+                IdleRead::Item(item) => item,
+                IdleRead::Ended => break,
+                IdleRead::IdleTimedOut => {
+                    // The upstream established the stream but sent no bytes for
+                    // `idle_timeout`. Rather than block on `next()` forever and
+                    // orphan the turn with no terminal event (#4761), surface a
+                    // terminal error so the consumer fires `chat_error`. Return
+                    // (not break) so the success `final_chunk()` below is NOT
+                    // emitted after the error — otherwise a stalled generation
+                    // would be reported as a normal `finish_reason: stop`.
+                    if let Some(dur) = idle_timeout {
+                        let _ = tx.send(Err(StreamError::IdleTimeout(dur))).await;
+                    }
+                    return;
+                }
+            };
             match item {
                 Ok(bytes) => {
                     // Convert bytes to string and process line by line
@@ -44,7 +106,10 @@ pub(crate) fn sse_bytes_to_chunks(
                                     e
                                 ))))
                                 .await;
-                            break;
+                            // Return (not break) so the success `final_chunk()`
+                            // below is not emitted after this terminal error —
+                            // only a clean EOF (`IdleRead::Ended`) reaches it.
+                            return;
                         }
                     };
 
@@ -74,12 +139,17 @@ pub(crate) fn sse_bytes_to_chunks(
                 }
                 Err(e) => {
                     let _ = tx.send(Err(StreamError::Http(e))).await;
-                    break;
+                    // Return (not break) so a transport failure is not masked by
+                    // the success `final_chunk()` below — only a clean EOF
+                    // (`IdleRead::Ended`) emits the terminal success chunk.
+                    return;
                 }
             }
         }
 
-        // Send final chunk
+        // Reached only on a clean EOF (`IdleRead::Ended`): every error path above
+        // returns after sending its `Err(...)`, so a success `final_chunk()` can
+        // never mask a decoding/transport/idle-timeout failure as a normal stop.
         let _ = tx.send(Ok(StreamChunk::final_chunk())).await;
     });
 
@@ -89,3 +159,51 @@ pub(crate) fn sse_bytes_to_chunks(
     })
     .boxed()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn idle_read_times_out_on_a_stalled_stream() {
+        // #4761: a stream that yields nothing (upstream established the
+        // connection then went silent mid-generation) must resolve to
+        // `IdleTimedOut` within the idle window instead of blocking forever.
+        // Without the timeout guard this `.await` never returns and the turn is
+        // orphaned with no terminal event.
+        let mut stalled = stream::pending::<u8>();
+        let outcome = read_next_or_idle(&mut stalled, Some(Duration::from_millis(20))).await;
+        assert_eq!(outcome, IdleRead::IdleTimedOut);
+    }
+
+    #[tokio::test]
+    async fn idle_read_returns_item_when_data_arrives() {
+        let mut s = stream::iter(vec![7u8, 8u8]);
+        assert_eq!(
+            read_next_or_idle(&mut s, Some(Duration::from_secs(5))).await,
+            IdleRead::Item(7u8)
+        );
+        assert_eq!(
+            read_next_or_idle(&mut s, Some(Duration::from_secs(5))).await,
+            IdleRead::Item(8u8)
+        );
+    }
+
+    #[tokio::test]
+    async fn idle_read_reports_clean_end() {
+        let mut empty = stream::iter(Vec::<u8>::new());
+        assert_eq!(
+            read_next_or_idle(&mut empty, Some(Duration::from_secs(5))).await,
+            IdleRead::Ended
+        );
+    }
+
+    #[tokio::test]
+    async fn idle_read_with_guard_disabled_still_yields_items() {
+        // `None` restores the pre-#4761 unbounded wait; a ready item must still
+        // be delivered (the disabled guard only removes the timeout, not data).
+        let mut s = stream::iter(vec![9u8]);
+        assert_eq!(read_next_or_idle(&mut s, None).await, IdleRead::Item(9u8));
+        assert_eq!(read_next_or_idle(&mut s, None).await, IdleRead::Ended);
+    }
+}

src/openhuman/inference/provider/compatible_timeout.rs

@@ -109,7 +109,7 @@ pub(super) fn connect_timeout() -> Duration {
 /// long response that keeps emitting tokens is never cut — only a genuine stall
 /// (no bytes from upstream, or a wedged consumer) for the whole window trips it.
 /// Resolved once per process and cached — see [`request_timeout`].
-pub(super) fn stream_idle_timeout() -> Duration {
+pub(crate) fn stream_idle_timeout() -> Duration {
     static CACHED: OnceLock<Duration> = OnceLock::new();
     *CACHED.get_or_init(|| {
         resolve(

src/openhuman/inference/provider/error_classify.rs

@@ -139,6 +139,12 @@ pub(crate) fn is_stream_error_non_retryable(err: &StreamError) -> bool {
         // JSON/SSE parse errors and IO errors are generally non-retryable
         StreamError::Json(_) | StreamError::InvalidSse(_) => true,
         StreamError::Io(_) => false,
+        // A mid-stream idle timeout (#4761) is a transient upstream stall — the
+        // connection went silent, not a permanent failure — so a retry on a
+        // fresh connection may succeed. If retries are exhausted it still
+        // surfaces as a terminal error (a `chat_error`), which is the whole
+        // point: the turn can no longer hang with no terminal event.
+        StreamError::IdleTimeout(_) => false,
     }
 }
 

src/openhuman/inference/provider/traits.rs

@@ -316,6 +316,11 @@ pub enum StreamError {
     #[error("Provider error: {0}")]
     Provider(String),
 
+    #[error(
+        "inference stream idle for {0:?} with no data — treating the upstream as stalled (#4761)"
+    )]
+    IdleTimeout(std::time::Duration),
+
     #[error("IO error: {0}")]
     Io(#[from] std::io::Error),
 }