From 887742a1e74e58ebc9989ea868d055a8358bb863 Mon Sep 17 00:00:00 2001
From: Chang Su <chang.s.su@oracle.com>
Date: Tue, 4 Nov 2025 15:38:43 -0800
Subject: [PATCH] [router][grpc] Fix index issues in reasoning content and
 missing streaming events (#12650)

---
 sgl-router/src/routers/grpc/harmony/streaming.rs      | 10 +++-------
 .../src/routers/grpc/regular/responses/handlers.rs    | 11 +++++++++++
 2 files changed, 14 insertions(+), 7 deletions(-)

diff --git a/sgl-router/src/routers/grpc/harmony/streaming.rs b/sgl-router/src/routers/grpc/harmony/streaming.rs
index 333ae8f03..6fa1e250b 100644
--- a/sgl-router/src/routers/grpc/harmony/streaming.rs
+++ b/sgl-router/src/routers/grpc/harmony/streaming.rs
@@ -615,7 +615,7 @@ impl HarmonyStreamingProcessor {
         let mut accumulated_tool_calls: Option<Vec<ToolCall>> = None;
 
         // Track which items we've started
-        let mut reasoning_output_index: Option<usize> = None;
+        let mut has_emitted_reasoning = false;
         let mut message_output_index: Option<usize> = None;
         let mut message_item_id: Option<String> = None;
         let mut has_emitted_content_part_added = false;
@@ -646,18 +646,14 @@ impl HarmonyStreamingProcessor {
                     if let Some(delta) = delta_result {
                         // Analysis channel → Reasoning item (wrapper events only, emitted once)
                         if let Some(_analysis_text) = &delta.analysis_delta {
-                            if reasoning_output_index.is_none() {
-                                // Allocate reasoning item and emit wrapper events
-                                let (output_index, _item_id) =
-                                    emitter.allocate_output_index(OutputItemType::Reasoning);
-                                reasoning_output_index = Some(output_index);
-
+                            if !has_emitted_reasoning {
                                 // Emit reasoning item (added + done in one call)
                                 // Note: reasoning_content will be provided at finalize
                                 emitter
                                     .emit_reasoning_item(tx, None)
                                     .map_err(|e| format!("Failed to emit reasoning item: {}", e))?;
 
+                                has_emitted_reasoning = true;
                                 has_analysis = true;
                             }
                         }
diff --git a/sgl-router/src/routers/grpc/regular/responses/handlers.rs b/sgl-router/src/routers/grpc/regular/responses/handlers.rs
index 5f82f4664..096d0208f 100644
--- a/sgl-router/src/routers/grpc/regular/responses/handlers.rs
+++ b/sgl-router/src/routers/grpc/regular/responses/handlers.rs
@@ -378,6 +378,17 @@ async fn process_and_transform_sse_stream(
     let created_at = chrono::Utc::now().timestamp() as u64;
     let mut event_emitter = ResponseStreamEventEmitter::new(response_id, model, created_at);
 
+    // Emit initial response.created and response.in_progress events
+    let event = event_emitter.emit_created();
+    event_emitter
+        .send_event(&event, &tx)
+        .map_err(|_| "Failed to send response.created event".to_string())?;
+
+    let event = event_emitter.emit_in_progress();
+    event_emitter
+        .send_event(&event, &tx)
+        .map_err(|_| "Failed to send response.in_progress event".to_string())?;
+
     // Convert body to data stream
     let mut stream = body.into_data_stream();