feat(completions): add Completion API streaming support to gRPC router

crates/protocols/src/completion.rs

@@ -262,6 +262,8 @@ pub struct CompletionStreamResponse {
     pub model: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub system_fingerprint: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub usage: Option<Usage>,
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize, schemars::JsonSchema)]

model_gateway/src/routers/grpc/pipeline.rs

@@ -428,6 +428,14 @@ impl RequestPipeline {
             None,
         );
 
+        let streaming_processor = Arc::new(streaming::StreamingProcessor::new(
+            ToolParserFactory::default(),
+            ReasoningParserFactory::default(),
+            None,
+            None,
+            metrics_labels::BACKEND_REGULAR,
+        ));
+
         let stages: Vec<Box<dyn PipelineStage>> = vec![
             Box::new(CompletionPreparationStage),
             Box::new(WorkerSelectionStage::new(
@@ -439,7 +447,10 @@ impl RequestPipeline {
             Box::new(CompletionRequestBuildingStage::new(false)), // No PD metadata
             Box::new(DispatchMetadataStage),
             Box::new(RequestExecutionStage::new(ExecutionMode::Single)),
-            Box::new(CompletionResponseProcessingStage::new(processor)),
+            Box::new(CompletionResponseProcessingStage::new(
+                processor,
+                streaming_processor,
+            )),
         ];
 
         Self {
@@ -460,6 +471,14 @@ impl RequestPipeline {
             None,
         );
 
+        let streaming_processor = Arc::new(streaming::StreamingProcessor::new(
+            ToolParserFactory::default(),
+            ReasoningParserFactory::default(),
+            None,
+            None,
+            metrics_labels::BACKEND_PD,
+        ));
+
         let stages: Vec<Box<dyn PipelineStage>> = vec![
             Box::new(CompletionPreparationStage),
             Box::new(WorkerSelectionStage::new(
@@ -471,7 +490,10 @@ impl RequestPipeline {
             Box::new(CompletionRequestBuildingStage::new(true)), // Inject PD metadata
             Box::new(DispatchMetadataStage),
             Box::new(RequestExecutionStage::new(ExecutionMode::DualDispatch)),
-            Box::new(CompletionResponseProcessingStage::new(processor)),
+            Box::new(CompletionResponseProcessingStage::new(
+                processor,
+                streaming_processor,
+            )),
         ];
 
         Self {

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

@@ -1,41 +1,52 @@
-//! Completion response processing stage (non-streaming)
+//! Completion response processing stage
 //!
-//! Stage 7 for the `/v1/completions` pipeline, parallel to
-//! `MessageResponseProcessingStage` from the Messages rollout.
-//! Converts backend `ProtoGenerateComplete` responses into OpenAI
-//! `CompletionResponse` format with `CompletionChoice` construction,
-//! `echo`/`suffix` handling, and legacy `LogProbs` formatting.
+//! Stage 7 for the `/v1/completions` pipeline
 //!
-//! Non-streaming only; streaming deferred to follow-up PR.
+//! - For streaming: spawns background task and returns SSE response (early exit)
+//! - For non-streaming: collects the backend response, converts it to
+//!   `CompletionResponse`, and stores it as `FinalResponse::Completion`.
 
-#![allow(dead_code)] //wired in pipeline factory follow-up PR
+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},
+        },
     },
 };
 
+/// Completion response processing stage
 pub(crate) struct CompletionResponseProcessingStage {
     processor: processor::ResponseProcessor,
+    streaming_processor: Arc<streaming::StreamingProcessor>,
 }
 
 impl CompletionResponseProcessingStage {
-    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 CompletionResponseProcessingStage {
     async fn execute(&self, ctx: &mut RequestContext) -> Result<Option<Response>, Response> {
+        let is_streaming = ctx.is_streaming();
+
         let execution_result = ctx.state.response.execution_result.take().ok_or_else(|| {
             error!(
                 function = "CompletionResponseProcessingStage::execute",
@@ -68,6 +79,26 @@ impl PipelineStage for CompletionResponseProcessingStage {
             )
         })?;
 
+        if is_streaming {
+            let response = self
+                .streaming_processor
+                .clone()
+                .process_completion_streaming_response(
+                    execution_result,
+                    ctx.completion_request_arc(),
+                    dispatch,
+                    tokenizer,
+                );
+
+            let response = match ctx.state.load_guards.take() {
+                Some(guards) => AttachedBody::wrap_response(response, guards),
+                None => response,
+            };
+
+            return Ok(Some(response));
+        }
+
+        // Non-streaming path
         let completion_request = ctx.completion_request_arc();
 
         let stop_decoder = ctx.state.response.stop_decoder.as_mut().ok_or_else(|| {