feat(gateway): add Messages API streaming support to gRPC router (#758)

Signed-off-by: Simo Lin <linsimo.mark@gmail.com>

Author: slin1237

model_gateway/src/routers/grpc/pipeline.rs

@@ -274,11 +274,19 @@ impl RequestPipeline {
         configured_reasoning_parser: Option<String>,
     ) -> Self {
         let processor = processor::ResponseProcessor::new(
+            tool_parser_factory.clone(),
+            reasoning_parser_factory.clone(),
+            configured_tool_parser.clone(),
+            configured_reasoning_parser.clone(),
+        );
+
+        let streaming_processor = Arc::new(streaming::StreamingProcessor::new(
             tool_parser_factory,
             reasoning_parser_factory,
             configured_tool_parser,
             configured_reasoning_parser,
-        );
+            metrics_labels::BACKEND_REGULAR,
+        ));
 
         let stages: Vec<Box<dyn PipelineStage>> = vec![
             Box::new(MessagePreparationStage),
@@ -291,7 +299,10 @@ impl RequestPipeline {
             Box::new(MessageRequestBuildingStage::new(false)), // No PD metadata
             Box::new(DispatchMetadataStage),
             Box::new(RequestExecutionStage::new(ExecutionMode::Single)),
-            Box::new(MessageResponseProcessingStage::new(processor)),
+            Box::new(MessageResponseProcessingStage::new(
+                processor,
+                streaming_processor,
+            )),
         ];
 
         Self {
@@ -310,11 +321,19 @@ impl RequestPipeline {
         configured_reasoning_parser: Option<String>,
     ) -> Self {
         let processor = processor::ResponseProcessor::new(
+            tool_parser_factory.clone(),
+            reasoning_parser_factory.clone(),
+            configured_tool_parser.clone(),
+            configured_reasoning_parser.clone(),
+        );
+
+        let streaming_processor = Arc::new(streaming::StreamingProcessor::new(
             tool_parser_factory,
             reasoning_parser_factory,
             configured_tool_parser,
             configured_reasoning_parser,
-        );
+            metrics_labels::BACKEND_PD,
+        ));
 
         let stages: Vec<Box<dyn PipelineStage>> = vec![
             Box::new(MessagePreparationStage),
@@ -327,7 +346,10 @@ impl RequestPipeline {
             Box::new(MessageRequestBuildingStage::new(true)), // Inject PD metadata
             Box::new(DispatchMetadataStage),
             Box::new(RequestExecutionStage::new(ExecutionMode::DualDispatch)),
-            Box::new(MessageResponseProcessingStage::new(processor)),
+            Box::new(MessageResponseProcessingStage::new(
+                processor,
+                streaming_processor,
+            )),
         ];
 
         Self {

model_gateway/src/routers/grpc/regular/processor.rs

@@ -487,13 +487,19 @@ impl ResponseProcessor {
         let complete = all_responses.into_iter().next().unwrap();
 
         // Check parser availability
-        // Always attempt reasoning parsing when a parser is available — some models'
-        // chat templates emit thinking tokens regardless of the request's `thinking` config.
-        let reasoning_parser_available = utils::check_reasoning_parser_availability(
-            &self.reasoning_parser_factory,
-            self.configured_reasoning_parser.as_deref(),
-            &messages_request.model,
+        // Only run reasoning parser when the user explicitly enabled thinking in the request.
+        // Without this gate, the reasoning parser misclassifies normal text and tool call JSON
+        // as thinking content, breaking tool use and producing incorrect content blocks.
+        let separate_reasoning = matches!(
+            &messages_request.thinking,
+            Some(messages::ThinkingConfig::Enabled { .. })
         );
+        let reasoning_parser_available = separate_reasoning
+            && utils::check_reasoning_parser_availability(
+                &self.reasoning_parser_factory,
+                self.configured_reasoning_parser.as_deref(),
+                &messages_request.model,
+            );
 
         let tool_choice_enabled = !matches!(
             &messages_request.tool_choice,
@@ -508,7 +514,7 @@ impl ResponseProcessor {
                 &messages_request.model,
             );
 
-        if !reasoning_parser_available {
+        if separate_reasoning && !reasoning_parser_available {
             tracing::debug!(
                 "No reasoning parser found for model '{}', reasoning content will not be separated",
                 messages_request.model

model_gateway/src/routers/grpc/regular/stages/messages/response_processing.rs

@@ -1,36 +1,50 @@
-//! Message response processing stage: non-streaming response processing
+//! Message response processing stage: streaming and non-streaming response processing
 //!
-//! Collects the backend response, converts it to an Anthropic `Message`,
-//! and stores it as FinalResponse::Messages.
-//! Streaming support will be added in a follow-up PR.
+//! - For streaming: Spawns background task and returns SSE response (early exit)
+//! - For non-streaming: Collects the backend response, converts it to an Anthropic `Message`,
+//!   and stores it as FinalResponse::Messages.
+
+use std::sync::Arc;
 
 use async_trait::async_trait;
 use axum::response::Response;
 use tracing::error;
 
-use crate::routers::{
-    error,
-    grpc::{
-        common::stages::PipelineStage,
-        context::{FinalResponse, RequestContext},
-        regular::processor,
+use crate::{
+    core::AttachedBody,
+    routers::{
+        error,
+        grpc::{
+            common::stages::PipelineStage,
+            context::{FinalResponse, RequestContext},
+            regular::{processor, streaming},
+        },
     },
 };
 
-/// Message response processing stage (non-streaming only)
+/// Message response processing stage
 pub(crate) struct MessageResponseProcessingStage {
     processor: processor::ResponseProcessor,
+    streaming_processor: Arc<streaming::StreamingProcessor>,
 }
 
 impl MessageResponseProcessingStage {
-    pub fn new(processor: processor::ResponseProcessor) -> Self {
-        Self { processor }
+    pub fn new(
+        processor: processor::ResponseProcessor,
+        streaming_processor: Arc<streaming::StreamingProcessor>,
+    ) -> Self {
+        Self {
+            processor,
+            streaming_processor,
+        }
     }
 }
 
 #[async_trait]
 impl PipelineStage for MessageResponseProcessingStage {
     async fn execute(&self, ctx: &mut RequestContext) -> Result<Option<Response>, Response> {
+        let is_streaming = ctx.is_streaming();
+
         // Extract execution result
         let execution_result = ctx.state.response.execution_result.take().ok_or_else(|| {
             error!(
@@ -66,6 +80,28 @@ impl PipelineStage for MessageResponseProcessingStage {
             )
         })?;
 
+        if is_streaming {
+            // Streaming: use StreamingProcessor and return SSE response
+            let response = self
+                .streaming_processor
+                .clone()
+                .process_messages_streaming_response(
+                    execution_result,
+                    ctx.messages_request_arc(),
+                    dispatch,
+                    tokenizer,
+                );
+
+            // Attach load guards for RAII lifecycle
+            let response = match ctx.state.load_guards.take() {
+                Some(guards) => AttachedBody::wrap_response(response, guards),
+                None => response,
+            };
+
+            return Ok(Some(response));
+        }
+
+        // Non-streaming: delegate to ResponseProcessor
         let messages_request = ctx.messages_request_arc();
 
         let stop_decoder = ctx.state.response.stop_decoder.as_mut().ok_or_else(|| {
@@ -79,7 +115,6 @@ impl PipelineStage for MessageResponseProcessingStage {
             )
         })?;
 
-        // Non-streaming: delegate to ResponseProcessor
         let response = self
             .processor
             .process_non_streaming_messages_response(