don't close stream of a response API observer

Author: Marlinski

shai-http/src/apis/openai/completion/handler.rs

@@ -62,7 +62,7 @@ async fn handle_chat_completion_stream(
     let formatter = ChatCompletionFormatter::new(model);
 
     // Create SSE stream
-    let stream = session_to_sse_stream(request_session, formatter, session_id);
+    let stream = session_to_sse_stream(request_session, formatter, session_id, true);
 
     Ok(Sse::new(stream).into_response())
 }

shai-http/src/apis/openai/response/handler.rs

@@ -69,7 +69,7 @@ async fn handle_response_stream(
     let formatter = ResponseFormatter::new(model, payload);
 
     // Create SSE stream
-    let stream = session_to_sse_stream(request_session, formatter, session_id);
+    let stream = session_to_sse_stream(request_session, formatter, session_id, true);
 
     Ok(Sse::new(stream).into_response())
 }
@@ -115,7 +115,8 @@ pub async fn handle_get_response(
     let formatter = ResponseFormatter::new(agent_session.agent_name.clone(), placeholder_payload);
 
     // Create SSE stream using the simple sse_stream (no lifecycle needed for read-only)
-    let stream = event_to_sse_stream(event_rx, formatter, response_id);
+    // stop_on_pause = false means stream stops on Completed OR Paused
+    let stream = event_to_sse_stream(event_rx, formatter, response_id, false);
 
     Ok(Sse::new(stream).into_response())
 }

shai-http/src/apis/simple/handler.rs

@@ -63,7 +63,7 @@ pub async fn handle_multimodal_query_stream(
     let formatter = SimpleFormatter::new(payload.model.clone());
 
     // Create SSE stream
-    let stream = session_to_sse_stream(request_session, formatter, session_id);
+    let stream = session_to_sse_stream(request_session, formatter, session_id, true);
 
     Ok(Sse::new(stream).into_response())
 }

shai-http/src/streaming.rs

@@ -36,6 +36,7 @@ fn sse_stream_internal<F, L>(
     formatter: F,
     session_id: String,
     lifecycle: Option<L>,
+    stop_on_pause: bool,
 ) -> impl Stream<Item = Result<Event, Infallible>>
 where
     F: EventFormatter + 'static,
@@ -55,7 +56,7 @@ where
                 loop {
                     match rx.next().await {
                         Some(Ok(event)) => {
-                            let is_terminal = is_terminal_event(&event);
+                            let is_terminal = is_terminal_event(&event, stop_on_pause);
                             let formatted = fmt.format_event(event, &session_id).await;
                             let new_done = if is_terminal { true } else { done };
 
@@ -93,23 +94,31 @@ where
 
 /// Core SSE stream creation from event receiver
 /// Watches events, formats them, and stops on completion or client disconnect
+///
+/// # Parameters
+/// * `stop_on_pause` - If true, only stops on Completed. If false, stops on Completed or StatusChanged to Paused.
 pub fn event_to_sse_stream<F>(
     event_rx: Receiver<AgentEvent>,
     formatter: F,
     session_id: String,
+    stop_on_pause: bool,
 ) -> impl Stream<Item = Result<Event, Infallible>>
 where
     F: EventFormatter + 'static,
 {
-    sse_stream_internal(event_rx, formatter, session_id, None::<()>)
+    sse_stream_internal(event_rx, formatter, session_id, None::<()>, stop_on_pause)
 }
 
 /// Create an SSE stream from a RequestSession
 /// Same as sse_stream but keeps lifecycle in scope for session cleanup
+///
+/// # Parameters
+/// * `stop_on_pause` - If true, only stops on Completed. If false, stops on Completed or StatusChanged to Paused.
 pub fn session_to_sse_stream<F>(
     request_session: RequestSession,
     formatter: F,
     session_id: String,
+    stop_on_pause: bool,
 ) -> impl Stream<Item = Result<Event, Infallible>>
 where
     F: EventFormatter + 'static,
@@ -118,17 +127,20 @@ where
     let _controller = request_session.controller;
     let lifecycle = request_session.lifecycle;
 
-    sse_stream_internal(event_rx, formatter, session_id, Some(lifecycle))
+    sse_stream_internal(event_rx, formatter, session_id, Some(lifecycle), stop_on_pause)
 }
 
 /// Check if an event signals the end of the stream
-fn is_terminal_event(event: &AgentEvent) -> bool {
-    matches!(
-        event,
-        AgentEvent::Completed { .. }
-            | AgentEvent::StatusChanged {
-                new_status: PublicAgentState::Paused,
-                ..
-            }
-    )
+///
+/// # Parameters
+/// * `stop_on_pause` - If true, only Completed is terminal. If false, both Completed and Paused are terminal.
+fn is_terminal_event(event: &AgentEvent, stop_on_pause: bool) -> bool {
+    match event {
+        AgentEvent::Completed { .. } => true,
+        AgentEvent::StatusChanged {
+            new_status: PublicAgentState::Paused,
+            ..
+        } => stop_on_pause,
+        _ => false,
+    }
 }