From f29d14b6d8539cd59be9607e4b70b027318b0952 Mon Sep 17 00:00:00 2001 From: VS Chandra Mourya Date: Fri, 13 Mar 2026 22:52:48 -0700 Subject: [PATCH 01/12] feat(protocol): validate /v1/completions request contract Add protocol-level validation for CompletionRequest and switch the /v1/completions HTTP handler to ValidatedJson so invalid requests fail before entering router-specific execution paths. Signed-off-by: VS Chandra Mourya --- crates/protocols/src/completion.rs | 99 +++++++++++++++++++++++++- model_gateway/src/server.rs | 2 +- model_gateway/tests/spec/completion.rs | 89 +++++++++++++++++++++++ 3 files changed, 187 insertions(+), 3 deletions(-) create mode 100644 model_gateway/tests/spec/completion.rs diff --git a/crates/protocols/src/completion.rs b/crates/protocols/src/completion.rs index e99c41fb88..db079c9620 100644 --- a/crates/protocols/src/completion.rs +++ b/crates/protocols/src/completion.rs @@ -2,35 +2,46 @@ use std::collections::HashMap; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; +use validator::Validate; -use super::common::*; +use super::{ + common::*, + sampling_params::{validate_top_k_value, validate_top_p_value}, +}; +use crate::validated::Normalizable; // ============================================================================ // Completions API (v1/completions) - DEPRECATED but still supported // ============================================================================ #[serde_with::skip_serializing_none] -#[derive(Debug, Clone, Deserialize, Serialize, schemars::JsonSchema)] +#[derive(Debug, Clone, Deserialize, Serialize, Validate, schemars::JsonSchema)] +#[validate(schema(function = "validate_completion_cross_parameters"))] pub struct CompletionRequest { /// ID of the model to use (required for OpenAI, optional for some implementations, such as SGLang) pub model: String, /// The prompt(s) to generate completions for + #[validate(custom(function = "validate_completion_prompt"))] pub prompt: StringOrArray, /// The suffix that comes after a completion of inserted text pub suffix: Option, /// The maximum number of tokens to generate + #[validate(range(min = 1))] pub max_tokens: Option, /// What sampling temperature to use, between 0 and 2 + #[validate(range(min = 0.0, max = 2.0))] pub temperature: Option, /// An alternative to sampling with temperature (nucleus sampling) + #[validate(custom(function = "validate_top_p_value"))] pub top_p: Option, /// How many completions to generate for each prompt + #[validate(range(min = 1, max = 10))] pub n: Option, /// Whether to stream back partial progress @@ -41,6 +52,7 @@ pub struct CompletionRequest { pub stream_options: Option, /// Include the log probabilities on the logprobs most likely tokens + #[validate(range(min = 0, max = 20))] pub logprobs: Option, /// Echo back the prompt in addition to the completion @@ -48,15 +60,19 @@ pub struct CompletionRequest { pub echo: bool, /// Up to 4 sequences where the API will stop generating further tokens + #[validate(custom(function = "validate_stop"))] pub stop: Option, /// Number between -2.0 and 2.0. Positive values penalize new tokens based on whether they appear in the text so far + #[validate(range(min = -2.0, max = 2.0))] pub presence_penalty: Option, /// Number between -2.0 and 2.0. Positive values penalize new tokens based on their existing frequency in the text so far + #[validate(range(min = -2.0, max = 2.0))] pub frequency_penalty: Option, /// Generates best_of completions server-side and returns the "best" + #[validate(range(min = 1))] pub best_of: Option, /// Modify the likelihood of specified tokens appearing in the completion @@ -70,15 +86,19 @@ pub struct CompletionRequest { // -------- Engine Specific Sampling Parameters -------- /// Top-k sampling parameter (-1 to disable) + #[validate(custom(function = "validate_top_k_value"))] pub top_k: Option, /// Min-p nucleus sampling parameter + #[validate(range(min = 0.0, max = 1.0))] pub min_p: Option, /// Minimum number of tokens to generate + #[validate(range(min = 1))] pub min_tokens: Option, /// Repetition penalty for reducing repetitive text + #[validate(range(min = 0.0, max = 2.0))] pub repetition_penalty: Option, /// Regex constraint for output generation @@ -123,6 +143,81 @@ pub struct CompletionRequest { pub other: Map, } +impl Normalizable for CompletionRequest {} + +fn validate_completion_prompt(prompt: &StringOrArray) -> Result<(), validator::ValidationError> { + match prompt { + StringOrArray::String(s) => { + if s.is_empty() { + let mut error = validator::ValidationError::new("prompt_empty"); + error.message = Some("prompt cannot be empty".into()); + return Err(error); + } + } + StringOrArray::Array(arr) => { + if arr.is_empty() { + let mut error = validator::ValidationError::new("prompt_empty"); + error.message = Some("prompt array cannot be empty".into()); + return Err(error); + } + + if arr.iter().any(|prompt| prompt.is_empty()) { + let mut error = validator::ValidationError::new("prompt_item_empty"); + error.message = Some("prompt array items cannot be empty".into()); + return Err(error); + } + } + } + + Ok(()) +} + +fn validate_completion_cross_parameters( + req: &CompletionRequest, +) -> Result<(), validator::ValidationError> { + if req.stream_options.is_some() && !req.stream { + let mut error = validator::ValidationError::new("stream_options_requires_stream"); + error.message = + Some("The 'stream_options' parameter is only allowed when 'stream' is enabled".into()); + return Err(error); + } + + if let (Some(min), Some(max)) = (req.min_tokens, req.max_tokens) { + if min > max { + let mut error = validator::ValidationError::new("min_tokens_exceeds_max"); + error.message = Some("min_tokens cannot exceed max_tokens".into()); + return Err(error); + } + } + + let constraint_count = [ + req.regex.is_some(), + req.ebnf.is_some(), + req.json_schema.is_some(), + ] + .iter() + .filter(|&&enabled| enabled) + .count(); + if constraint_count > 1 { + let mut error = validator::ValidationError::new("multiple_constraints"); + error.message = Some( + "only one structured output constraint (regex, ebnf, or json_schema) can be active at a time" + .into(), + ); + return Err(error); + } + + if let (Some(best_of), Some(n)) = (req.best_of, req.n) { + if best_of < n { + let mut error = validator::ValidationError::new("best_of_less_than_n"); + error.message = Some("best_of must be greater than or equal to n".into()); + return Err(error); + } + } + + Ok(()) +} + impl GenerationRequest for CompletionRequest { fn is_stream(&self) -> bool { self.stream diff --git a/model_gateway/src/server.rs b/model_gateway/src/server.rs index e17d5f4c54..0d0fd79452 100644 --- a/model_gateway/src/server.rs +++ b/model_gateway/src/server.rs @@ -201,7 +201,7 @@ async fn v1_chat_completions( async fn v1_completions( State(state): State>, headers: http::HeaderMap, - Json(body): Json, + ValidatedJson(body): ValidatedJson, ) -> Response { state .router diff --git a/model_gateway/tests/spec/completion.rs b/model_gateway/tests/spec/completion.rs new file mode 100644 index 0000000000..e3894d16fc --- /dev/null +++ b/model_gateway/tests/spec/completion.rs @@ -0,0 +1,89 @@ +use openai_protocol::{ + common::{StreamOptions, StringOrArray}, + completion::CompletionRequest, + validated::Normalizable, +}; +use serde_json::{Map, Value}; +use validator::Validate; + +fn base_request() -> CompletionRequest { + CompletionRequest { + model: "test-model".to_string(), + prompt: StringOrArray::String("hello".to_string()), + suffix: None, + max_tokens: Some(16), + temperature: None, + top_p: None, + n: None, + stream: false, + stream_options: None, + logprobs: None, + echo: false, + stop: None, + presence_penalty: None, + frequency_penalty: None, + best_of: None, + logit_bias: None, + user: None, + seed: None, + top_k: None, + min_p: None, + min_tokens: None, + repetition_penalty: None, + regex: None, + ebnf: None, + json_schema: None, + stop_token_ids: None, + no_stop_trim: false, + ignore_eos: false, + skip_special_tokens: true, + lora_path: None, + session_params: None, + return_hidden_states: false, + sampling_seed: None, + other: Map::::new(), + } +} + +#[test] +fn test_completion_prompt_cannot_be_empty() { + let mut req = base_request(); + req.prompt = StringOrArray::String(String::new()); + req.normalize(); + assert!(req.validate().is_err()); +} + +#[test] +fn test_completion_prompt_array_items_cannot_be_empty() { + let mut req = base_request(); + req.prompt = StringOrArray::Array(vec!["first".to_string(), String::new()]); + req.normalize(); + assert!(req.validate().is_err()); +} + +#[test] +fn test_completion_prompt_array_is_protocol_valid() { + let mut req = base_request(); + req.prompt = StringOrArray::Array(vec!["first".to_string(), "second".to_string()]); + req.normalize(); + assert!(req.validate().is_ok()); +} + +#[test] +fn test_completion_stream_options_require_stream() { + let mut req = base_request(); + req.stream_options = Some(StreamOptions { + include_usage: Some(true), + }); + req.normalize(); + assert!(req.validate().is_err()); +} + +#[test] +fn test_completion_best_of_must_cover_n() { + let mut req = base_request(); + req.n = Some(3); + req.best_of = Some(2); + req.normalize(); + assert!(req.validate().is_err()); +} From 38170bc8965204f7ae8405f7cb9c0eaa0d945c38 Mon Sep 17 00:00:00 2001 From: VS Chandra Mourya Date: Wed, 18 Mar 2026 23:30:20 -0700 Subject: [PATCH 02/12] fix(tests) remove test completion file Signed-off-by: VS Chandra Mourya --- model_gateway/tests/spec/completion.rs | 89 -------------------------- 1 file changed, 89 deletions(-) delete mode 100644 model_gateway/tests/spec/completion.rs diff --git a/model_gateway/tests/spec/completion.rs b/model_gateway/tests/spec/completion.rs deleted file mode 100644 index e3894d16fc..0000000000 --- a/model_gateway/tests/spec/completion.rs +++ /dev/null @@ -1,89 +0,0 @@ -use openai_protocol::{ - common::{StreamOptions, StringOrArray}, - completion::CompletionRequest, - validated::Normalizable, -}; -use serde_json::{Map, Value}; -use validator::Validate; - -fn base_request() -> CompletionRequest { - CompletionRequest { - model: "test-model".to_string(), - prompt: StringOrArray::String("hello".to_string()), - suffix: None, - max_tokens: Some(16), - temperature: None, - top_p: None, - n: None, - stream: false, - stream_options: None, - logprobs: None, - echo: false, - stop: None, - presence_penalty: None, - frequency_penalty: None, - best_of: None, - logit_bias: None, - user: None, - seed: None, - top_k: None, - min_p: None, - min_tokens: None, - repetition_penalty: None, - regex: None, - ebnf: None, - json_schema: None, - stop_token_ids: None, - no_stop_trim: false, - ignore_eos: false, - skip_special_tokens: true, - lora_path: None, - session_params: None, - return_hidden_states: false, - sampling_seed: None, - other: Map::::new(), - } -} - -#[test] -fn test_completion_prompt_cannot_be_empty() { - let mut req = base_request(); - req.prompt = StringOrArray::String(String::new()); - req.normalize(); - assert!(req.validate().is_err()); -} - -#[test] -fn test_completion_prompt_array_items_cannot_be_empty() { - let mut req = base_request(); - req.prompt = StringOrArray::Array(vec!["first".to_string(), String::new()]); - req.normalize(); - assert!(req.validate().is_err()); -} - -#[test] -fn test_completion_prompt_array_is_protocol_valid() { - let mut req = base_request(); - req.prompt = StringOrArray::Array(vec!["first".to_string(), "second".to_string()]); - req.normalize(); - assert!(req.validate().is_ok()); -} - -#[test] -fn test_completion_stream_options_require_stream() { - let mut req = base_request(); - req.stream_options = Some(StreamOptions { - include_usage: Some(true), - }); - req.normalize(); - assert!(req.validate().is_err()); -} - -#[test] -fn test_completion_best_of_must_cover_n() { - let mut req = base_request(); - req.n = Some(3); - req.best_of = Some(2); - req.normalize(); - assert!(req.validate().is_err()); -} From a426df106caf981e73ec052012ac37b255cfc810 Mon Sep 17 00:00:00 2001 From: VS Chandra Mourya Date: Wed, 18 Mar 2026 23:30:40 -0700 Subject: [PATCH 03/12] fix(protocol): align completion request contract with OpenAI Signed-off-by: VS Chandra Mourya --- crates/protocols/src/completion.rs | 88 ++++++++++++++---------------- 1 file changed, 41 insertions(+), 47 deletions(-) diff --git a/crates/protocols/src/completion.rs b/crates/protocols/src/completion.rs index db079c9620..46ab314987 100644 --- a/crates/protocols/src/completion.rs +++ b/crates/protocols/src/completion.rs @@ -25,66 +25,66 @@ pub struct CompletionRequest { #[validate(custom(function = "validate_completion_prompt"))] pub prompt: StringOrArray, - /// The suffix that comes after a completion of inserted text - pub suffix: Option, + /// Generates `best_of` completions server-side and returns the "best" + #[validate(range(min = 0, max = 20))] + pub best_of: Option, + + /// Echo back the prompt in addition to the completion + #[serde(default)] + pub echo: bool, + + /// Number between -2.0 and 2.0. Positive values penalize new tokens based on their existing frequency in the text so far + #[validate(range(min = -2.0, max = 2.0))] + pub frequency_penalty: Option, + + /// Modify the likelihood of specified tokens appearing in the completion + pub logit_bias: Option>, + + /// Include the log probabilities on the `logprobs` most likely tokens + #[validate(range(min = 0, max = 5))] + pub logprobs: Option, /// The maximum number of tokens to generate #[validate(range(min = 1))] pub max_tokens: Option, - /// What sampling temperature to use, between 0 and 2 - #[validate(range(min = 0.0, max = 2.0))] - pub temperature: Option, - - /// An alternative to sampling with temperature (nucleus sampling) - #[validate(custom(function = "validate_top_p_value"))] - pub top_p: Option, - /// How many completions to generate for each prompt - #[validate(range(min = 1, max = 10))] + #[validate(range(min = 1, max = 128))] pub n: Option, - /// Whether to stream back partial progress - #[serde(default)] - pub stream: bool, - - /// Options for streaming response - pub stream_options: Option, - - /// Include the log probabilities on the logprobs most likely tokens - #[validate(range(min = 0, max = 20))] - pub logprobs: Option, + /// Number between -2.0 and 2.0. Positive values penalize new tokens based on whether they appear in the text so far + #[validate(range(min = -2.0, max = 2.0))] + pub presence_penalty: Option, - /// Echo back the prompt in addition to the completion - #[serde(default)] - pub echo: bool, + /// If specified, our system will make a best effort to sample deterministically + pub seed: Option, /// Up to 4 sequences where the API will stop generating further tokens #[validate(custom(function = "validate_stop"))] pub stop: Option, - /// Number between -2.0 and 2.0. Positive values penalize new tokens based on whether they appear in the text so far - #[validate(range(min = -2.0, max = 2.0))] - pub presence_penalty: Option, + /// Whether to stream back partial progress + #[serde(default)] + pub stream: bool, - /// Number between -2.0 and 2.0. Positive values penalize new tokens based on their existing frequency in the text so far - #[validate(range(min = -2.0, max = 2.0))] - pub frequency_penalty: Option, + /// Options for streaming response + pub stream_options: Option, - /// Generates best_of completions server-side and returns the "best" - #[validate(range(min = 1))] - pub best_of: Option, + /// The suffix that comes after a completion of inserted text + pub suffix: Option, - /// Modify the likelihood of specified tokens appearing in the completion - pub logit_bias: Option>, + /// What sampling temperature to use, between 0 and 2 + #[validate(range(min = 0.0, max = 2.0))] + pub temperature: Option, + + /// An alternative to sampling with temperature (nucleus sampling) + #[validate(custom(function = "validate_top_p_value"))] + pub top_p: Option, /// A unique identifier representing your end-user pub user: Option, - /// If specified, our system will make a best effort to sample deterministically - pub seed: Option, - - // -------- Engine Specific Sampling Parameters -------- + // Engine-specific sampling parameters /// Top-k sampling parameter (-1 to disable) #[validate(custom(function = "validate_top_k_value"))] pub top_k: Option, @@ -190,14 +190,8 @@ fn validate_completion_cross_parameters( } } - let constraint_count = [ - req.regex.is_some(), - req.ebnf.is_some(), - req.json_schema.is_some(), - ] - .iter() - .filter(|&&enabled| enabled) - .count(); + let constraint_count = + req.regex.is_some() as u8 + req.ebnf.is_some() as u8 + req.json_schema.is_some() as u8; if constraint_count > 1 { let mut error = validator::ValidationError::new("multiple_constraints"); error.message = Some( From a402f0f5b3ba8aa9316cfb7a7f44548ee0e172ad Mon Sep 17 00:00:00 2001 From: VS Chandra Mourya Date: Wed, 18 Mar 2026 23:43:56 -0700 Subject: [PATCH 04/12] Include codex suggestions Signed-off-by: VS Chandra Mourya --- crates/protocols/src/completion.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/protocols/src/completion.rs b/crates/protocols/src/completion.rs index 46ab314987..b037e790f6 100644 --- a/crates/protocols/src/completion.rs +++ b/crates/protocols/src/completion.rs @@ -45,7 +45,7 @@ pub struct CompletionRequest { pub logprobs: Option, /// The maximum number of tokens to generate - #[validate(range(min = 1))] + #[validate(range(min = 0))] pub max_tokens: Option, /// How many completions to generate for each prompt @@ -202,9 +202,9 @@ fn validate_completion_cross_parameters( } if let (Some(best_of), Some(n)) = (req.best_of, req.n) { - if best_of < n { + if best_of <= n { let mut error = validator::ValidationError::new("best_of_less_than_n"); - error.message = Some("best_of must be greater than or equal to n".into()); + error.message = Some("best_of must be greater than n".into()); return Err(error); } } From 6e4beb63b6be8debbafef144b0fade56f4a8d720 Mon Sep 17 00:00:00 2001 From: VS Chandra Mourya Date: Fri, 20 Mar 2026 14:04:13 -0700 Subject: [PATCH 05/12] fix(protocol): align completion validation with OpenAI Signed-off-by: VS Chandra Mourya --- crates/protocols/src/completion.rs | 20 +++++++------------- 1 file changed, 7 insertions(+), 13 deletions(-) diff --git a/crates/protocols/src/completion.rs b/crates/protocols/src/completion.rs index b037e790f6..01b1adcf9e 100644 --- a/crates/protocols/src/completion.rs +++ b/crates/protocols/src/completion.rs @@ -147,25 +147,13 @@ impl Normalizable for CompletionRequest {} fn validate_completion_prompt(prompt: &StringOrArray) -> Result<(), validator::ValidationError> { match prompt { - StringOrArray::String(s) => { - if s.is_empty() { - let mut error = validator::ValidationError::new("prompt_empty"); - error.message = Some("prompt cannot be empty".into()); - return Err(error); - } - } + StringOrArray::String(_) => {} StringOrArray::Array(arr) => { if arr.is_empty() { let mut error = validator::ValidationError::new("prompt_empty"); error.message = Some("prompt array cannot be empty".into()); return Err(error); } - - if arr.iter().any(|prompt| prompt.is_empty()) { - let mut error = validator::ValidationError::new("prompt_item_empty"); - error.message = Some("prompt array items cannot be empty".into()); - return Err(error); - } } } @@ -209,6 +197,12 @@ fn validate_completion_cross_parameters( } } + if req.stream && req.best_of.is_some() { + let mut error = validator::ValidationError::new("best_of_not_supported_with_stream"); + error.message = Some("best_of is not supported when stream is enabled".into()); + return Err(error); + } + Ok(()) } From 82af9a9e99a85f99f6464319064afdef8839f2a2 Mon Sep 17 00:00:00 2001 From: VS Chandra Mourya Date: Sun, 15 Mar 2026 17:16:08 -0700 Subject: [PATCH 06/12] feat(completions): add native gRPC pipeline typing for /v1/completions Introduce CompletionRequest as a first-class gRPC pipeline request type and extend shared pipeline orchestration to carry completion requests and completion responses natively. Signed-off-by: VS Chandra Mourya --- .../grpc/common/stages/dispatch_metadata.rs | 1 + model_gateway/src/routers/grpc/context.rs | 60 +++++++++ .../grpc/harmony/stages/request_building.rs | 52 ++++++-- .../harmony/stages/response_processing.rs | 52 ++++++-- model_gateway/src/routers/grpc/pipeline.rs | 120 +++++++++++++++++- .../grpc/regular/stages/request_building.rs | 4 +- .../regular/stages/response_processing.rs | 4 +- 7 files changed, 273 insertions(+), 20 deletions(-) diff --git a/model_gateway/src/routers/grpc/common/stages/dispatch_metadata.rs b/model_gateway/src/routers/grpc/common/stages/dispatch_metadata.rs index 4e5aeead43..bb0cc3ebef 100644 --- a/model_gateway/src/routers/grpc/common/stages/dispatch_metadata.rs +++ b/model_gateway/src/routers/grpc/common/stages/dispatch_metadata.rs @@ -32,6 +32,7 @@ impl PipelineStage for DispatchMetadataStage { let request_id = proto_request.request_id().to_string(); let model = match &ctx.input.request_type { RequestType::Chat(req) => req.model.clone(), + RequestType::Completion(req) => req.model.clone(), RequestType::Generate(_req) => { // Generate requests don't have a model field // Use model_id from input or UNKNOWN_MODEL_ID diff --git a/model_gateway/src/routers/grpc/context.rs b/model_gateway/src/routers/grpc/context.rs index 54a794af43..2c0f559d44 100644 --- a/model_gateway/src/routers/grpc/context.rs +++ b/model_gateway/src/routers/grpc/context.rs @@ -11,6 +11,7 @@ use llm_tokenizer::{stop::StopSequenceDecoder, traits::Tokenizer, TokenizerRegis use openai_protocol::{ chat::{ChatCompletionRequest, ChatCompletionResponse}, classify::{ClassifyRequest, ClassifyResponse}, + completion::{CompletionRequest, CompletionResponse}, embedding::{EmbeddingRequest, EmbeddingResponse}, generate::{GenerateRequest, GenerateResponse}, messages::{CreateMessageRequest, Message}, @@ -50,6 +51,7 @@ pub(crate) struct RequestInput { pub(crate) enum RequestType { Chat(Arc), Generate(Arc), + Completion(Arc), Responses(Arc), Embedding(Arc), Classify(Arc), @@ -61,6 +63,7 @@ impl std::fmt::Display for RequestType { match self { Self::Chat(_) => write!(f, "Chat"), Self::Generate(_) => write!(f, "Generate"), + Self::Completion(_) => write!(f, "Completion"), Self::Responses(_) => write!(f, "Responses"), Self::Embedding(_) => write!(f, "Embedding"), Self::Classify(_) => write!(f, "Classify"), @@ -254,6 +257,24 @@ impl RequestContext { } } + /// Create context for completion request + pub fn for_completion( + request: Arc, + headers: Option, + model_id: Option, + components: Arc, + ) -> Self { + Self { + input: RequestInput { + request_type: RequestType::Completion(request), + headers, + model_id, + }, + components, + state: ProcessingState::default(), + } + } + /// Create context for Responses API request pub fn for_responses( request: Arc, @@ -374,6 +395,38 @@ impl RequestContext { } } + /// Get completion request (panics if not completion) + #[expect( + dead_code, + reason = "ref accessor provided for API completeness alongside Arc accessor" + )] + #[expect( + clippy::panic, + reason = "typed accessor: caller guarantees variant via RequestType construction" + )] + pub fn completion_request(&self) -> &CompletionRequest { + match &self.input.request_type { + RequestType::Completion(req) => req.as_ref(), + _ => panic!("Expected completion request"), + } + } + + /// Get Arc clone of completion request (panics if not completion) + #[expect( + dead_code, + reason = "Arc accessor is introduced before later stacked PRs use it from completion stages" + )] + #[expect( + clippy::panic, + reason = "typed accessor: caller guarantees variant via RequestType construction" + )] + pub fn completion_request_arc(&self) -> Arc { + match &self.input.request_type { + RequestType::Completion(req) => Arc::clone(req), + _ => panic!("Expected completion request"), + } + } + /// Get Arc clone of responses request (panics if not responses) #[expect( clippy::panic, @@ -419,6 +472,7 @@ impl RequestContext { match &self.input.request_type { RequestType::Chat(req) => req.stream, RequestType::Generate(req) => req.stream, + RequestType::Completion(req) => req.stream, RequestType::Responses(req) => req.stream.unwrap_or(false), RequestType::Messages(req) => req.stream.unwrap_or(false), RequestType::Embedding(_) => false, // Embeddings are never streaming @@ -597,10 +651,16 @@ pub(crate) enum ExecutionResult { /// Final processed response #[derive(Debug)] +#[expect( + dead_code, + reason = "Completion final response is introduced in the plumbing PR before later stacked PRs produce it" +)] pub(crate) enum FinalResponse { Chat(ChatCompletionResponse), /// Generate response is a Vec of GenerateResponse (n=1 returns single item, n>1 returns multiple) Generate(Vec), + /// Completion response (OpenAI /v1/completions format) + Completion(CompletionResponse), /// Embedding response Embedding(EmbeddingResponse), /// Classification response diff --git a/model_gateway/src/routers/grpc/harmony/stages/request_building.rs b/model_gateway/src/routers/grpc/harmony/stages/request_building.rs index d8e97505b2..eeca3cb92e 100644 --- a/model_gateway/src/routers/grpc/harmony/stages/request_building.rs +++ b/model_gateway/src/routers/grpc/harmony/stages/request_building.rs @@ -62,18 +62,54 @@ impl PipelineStage for HarmonyRequestBuildingStage { let request_id = match &ctx.input.request_type { RequestType::Chat(_) => format!("chatcmpl-{}", Uuid::now_v7()), RequestType::Responses(_) => format!("responses-{}", Uuid::now_v7()), - request_type @ (RequestType::Generate(_) - | RequestType::Embedding(_) - | RequestType::Classify(_) - | RequestType::Messages(_)) => { + RequestType::Generate(_) => { error!( function = "HarmonyRequestBuildingStage::execute", - request_type = %request_type, - "{request_type} request type not supported for Harmony models" + "Generate request type not supported for Harmony models" ); return Err(error::bad_request( - "not_supported_in_harmony", - format!("{request_type} requests are not supported with Harmony models"), + "harmony_generate_not_supported", + "Generate requests are not supported with Harmony models".to_string(), + )); + } + RequestType::Completion(_) => { + error!( + function = "HarmonyRequestBuildingStage::execute", + "Completion request type not supported for Harmony models" + ); + return Err(error::bad_request( + "harmony_completion_not_supported", + "Completion requests are not supported with Harmony models".to_string(), + )); + } + RequestType::Embedding(_) => { + error!( + function = "HarmonyRequestBuildingStage::execute", + "Embedding request type not supported for Harmony models" + ); + return Err(error::bad_request( + "harmony_embedding_not_supported", + "Embedding requests are not supported with Harmony models".to_string(), + )); + } + RequestType::Classify(_) => { + error!( + function = "HarmonyRequestBuildingStage::execute", + "Classify request type not supported for Harmony models" + ); + return Err(error::bad_request( + "harmony_classify_not_supported", + "Classify requests are not supported with Harmony models".to_string(), + )); + } + RequestType::Messages(_) => { + error!( + function = "HarmonyRequestBuildingStage::execute", + "Messages request type not supported for Harmony models" + ); + return Err(error::bad_request( + "harmony_messages_not_supported", + "Messages requests are not supported with Harmony models".to_string(), )); } }; diff --git a/model_gateway/src/routers/grpc/harmony/stages/response_processing.rs b/model_gateway/src/routers/grpc/harmony/stages/response_processing.rs index 6a6d6961eb..4bb14ddfde 100644 --- a/model_gateway/src/routers/grpc/harmony/stages/response_processing.rs +++ b/model_gateway/src/routers/grpc/harmony/stages/response_processing.rs @@ -138,18 +138,54 @@ impl PipelineStage for HarmonyResponseProcessingStage { ctx.state.response.responses_iteration_result = Some(iteration_result); Ok(None) } - request_type @ (RequestType::Generate(_) - | RequestType::Embedding(_) - | RequestType::Classify(_) - | RequestType::Messages(_)) => { + RequestType::Generate(_) => { error!( function = "HarmonyResponseProcessingStage::execute", - request_type = %request_type, - "{request_type} request type not supported in Harmony pipeline" + "Generate request type not supported in Harmony pipeline" ); Err(error::internal_error( - "not_supported_in_harmony", - format!("{request_type} requests not supported in Harmony pipeline"), + "harmony_generate_not_supported", + "Generate requests not supported in Harmony pipeline", + )) + } + RequestType::Completion(_) => { + error!( + function = "HarmonyResponseProcessingStage::execute", + "Completion request type not supported in Harmony pipeline" + ); + Err(error::internal_error( + "harmony_completion_not_supported", + "Completion requests not supported in Harmony pipeline", + )) + } + RequestType::Embedding(_) => { + error!( + function = "HarmonyResponseProcessingStage::execute", + "Embedding request type not supported in Harmony pipeline" + ); + Err(error::internal_error( + "harmony_embedding_not_supported", + "Embedding requests not supported in Harmony pipeline", + )) + } + RequestType::Classify(_) => { + error!( + function = "HarmonyResponseProcessingStage::execute", + "Classify request type not supported in Harmony pipeline" + ); + Err(error::internal_error( + "harmony_classify_not_supported", + "Classify requests not supported in Harmony pipeline", + )) + } + RequestType::Messages(_) => { + error!( + function = "HarmonyResponseProcessingStage::execute", + "Messages request type not supported in Harmony pipeline" + ); + Err(error::internal_error( + "harmony_messages_not_supported", + "Messages requests not supported in Harmony pipeline", )) } } diff --git a/model_gateway/src/routers/grpc/pipeline.rs b/model_gateway/src/routers/grpc/pipeline.rs index 41b801ddf2..a1f7ca22a2 100644 --- a/model_gateway/src/routers/grpc/pipeline.rs +++ b/model_gateway/src/routers/grpc/pipeline.rs @@ -9,6 +9,7 @@ use axum::response::{IntoResponse, Response}; use openai_protocol::{ chat::{ChatCompletionRequest, ChatCompletionResponse}, classify::ClassifyRequest, + completion::CompletionRequest, embedding::EmbeddingRequest, generate::GenerateRequest, messages::CreateMessageRequest, @@ -430,12 +431,13 @@ impl RequestPipeline { axum::Json(response).into_response() } Some(FinalResponse::Generate(_)) + | Some(FinalResponse::Completion(_)) | Some(FinalResponse::Embedding(_)) | Some(FinalResponse::Classify(_)) | Some(FinalResponse::Messages(_)) => { error!( function = "execute_chat", - "Wrong response type: expected Chat, got Generate/Embedding/Classify/Messages" + "Wrong response type: expected Chat, got Generate/Completion/Embedding/Classify/Messages" ); Metrics::record_router_error( metrics_labels::ROUTER_GRPC, @@ -465,6 +467,7 @@ impl RequestPipeline { } } + /// Execute the complete pipeline for a generate request /// Execute the complete pipeline for a generate request pub async fn execute_generate( &self, @@ -534,12 +537,13 @@ impl RequestPipeline { axum::Json(response).into_response() } Some(FinalResponse::Chat(_)) + | Some(FinalResponse::Completion(_)) | Some(FinalResponse::Embedding(_)) | Some(FinalResponse::Classify(_)) | Some(FinalResponse::Messages(_)) => { error!( function = "execute_generate", - "Wrong response type: expected Generate, got Chat/Embedding/Classify/Messages" + "Wrong response type: expected Generate, got Chat/Completion/Embedding/Classify/Messages" ); Metrics::record_router_error( metrics_labels::ROUTER_GRPC, @@ -569,6 +573,116 @@ impl RequestPipeline { } } + /// Execute the complete pipeline for a completion request + #[expect( + dead_code, + reason = "Completion pipeline entrypoint is introduced before later stacked PRs wire the router to call it" + )] + pub async fn execute_completion( + &self, + request: Arc, + headers: Option, + model_id: Option, + components: Arc, + ) -> Response { + let start = Instant::now(); + let model = request.model.clone(); + let streaming = request.stream; + + Metrics::record_router_request( + metrics_labels::ROUTER_GRPC, + self.backend_type, + metrics_labels::CONNECTION_GRPC, + &model, + metrics_labels::ENDPOINT_COMPLETIONS, + bool_to_static_str(streaming), + ); + + let mut ctx = + RequestContext::for_completion(request, headers, model_id.clone(), components); + + for stage in self.stages.iter() { + match stage.execute(&mut ctx).await { + Ok(Some(response)) => { + Metrics::record_router_duration( + metrics_labels::ROUTER_GRPC, + self.backend_type, + metrics_labels::CONNECTION_GRPC, + &model, + metrics_labels::ENDPOINT_COMPLETIONS, + start.elapsed(), + ); + return response; + } + Ok(None) => continue, + Err(response) => { + Metrics::record_router_error( + metrics_labels::ROUTER_GRPC, + self.backend_type, + metrics_labels::CONNECTION_GRPC, + &model, + metrics_labels::ENDPOINT_COMPLETIONS, + error_type_from_status(response.status()), + ); + error!( + "Stage {} failed with status {}", + stage.name(), + response.status() + ); + return response; + } + } + } + + match ctx.state.response.final_response { + Some(FinalResponse::Completion(response)) => { + Metrics::record_router_duration( + metrics_labels::ROUTER_GRPC, + self.backend_type, + metrics_labels::CONNECTION_GRPC, + &model, + metrics_labels::ENDPOINT_COMPLETIONS, + start.elapsed(), + ); + axum::Json(response).into_response() + } + Some(FinalResponse::Chat(_)) + | Some(FinalResponse::Generate(_)) + | Some(FinalResponse::Embedding(_)) + | Some(FinalResponse::Classify(_)) + | Some(FinalResponse::Messages(_)) => { + error!( + function = "execute_completion", + "Wrong response type: expected Completion" + ); + Metrics::record_router_error( + metrics_labels::ROUTER_GRPC, + self.backend_type, + metrics_labels::CONNECTION_GRPC, + &model, + metrics_labels::ENDPOINT_COMPLETIONS, + metrics_labels::ERROR_INTERNAL, + ); + error::internal_error("wrong_response_type", "Internal error: wrong response type") + } + None => { + error!( + function = "execute_completion", + "No response produced by pipeline" + ); + Metrics::record_router_error( + metrics_labels::ROUTER_GRPC, + self.backend_type, + metrics_labels::CONNECTION_GRPC, + &model, + metrics_labels::ENDPOINT_COMPLETIONS, + metrics_labels::ERROR_INTERNAL, + ); + error::internal_error("no_response_produced", "No response produced") + } + } + } + /// Execute the complete pipeline for an embedding request pub async fn execute_embeddings( &self, @@ -840,6 +954,7 @@ impl RequestPipeline { } Some(FinalResponse::Chat(_)) | Some(FinalResponse::Generate(_)) + | Some(FinalResponse::Completion(_)) | Some(FinalResponse::Embedding(_)) | Some(FinalResponse::Classify(_)) => { error!( @@ -921,6 +1036,7 @@ impl RequestPipeline { match ctx.state.response.final_response { Some(FinalResponse::Chat(response)) => Ok(response), Some(FinalResponse::Generate(_)) + | Some(FinalResponse::Completion(_)) | Some(FinalResponse::Embedding(_)) | Some(FinalResponse::Classify(_)) | Some(FinalResponse::Messages(_)) => { diff --git a/model_gateway/src/routers/grpc/regular/stages/request_building.rs b/model_gateway/src/routers/grpc/regular/stages/request_building.rs index c478963cae..397502593b 100644 --- a/model_gateway/src/routers/grpc/regular/stages/request_building.rs +++ b/model_gateway/src/routers/grpc/regular/stages/request_building.rs @@ -41,7 +41,9 @@ impl PipelineStage for RequestBuildingStage { RequestType::Generate(_) => self.generate_stage.execute(ctx).await, RequestType::Embedding(_) => self.embedding_stage.execute(ctx).await, RequestType::Classify(_) => self.embedding_stage.execute(ctx).await, - request_type @ (RequestType::Responses(_) | RequestType::Messages(_)) => { + request_type @ (RequestType::Completion(_) + | RequestType::Responses(_) + | RequestType::Messages(_)) => { error!( function = "RequestBuildingStage::execute", request_type = %request_type, diff --git a/model_gateway/src/routers/grpc/regular/stages/response_processing.rs b/model_gateway/src/routers/grpc/regular/stages/response_processing.rs index 2bd6f51a5f..88ab6408e6 100644 --- a/model_gateway/src/routers/grpc/regular/stages/response_processing.rs +++ b/model_gateway/src/routers/grpc/regular/stages/response_processing.rs @@ -53,7 +53,9 @@ impl PipelineStage for ResponseProcessingStage { RequestType::Generate(_) => self.generate_stage.execute(ctx).await, RequestType::Embedding(_) => self.embedding_stage.execute(ctx).await, RequestType::Classify(_) => self.classify_stage.execute(ctx).await, - request_type @ (RequestType::Responses(_) | RequestType::Messages(_)) => { + request_type @ (RequestType::Completion(_) + | RequestType::Responses(_) + | RequestType::Messages(_)) => { error!( function = "ResponseProcessingStage::execute", request_type = %request_type, From 8e3fefdd38ad67fa55fe9bb665170fcdab3b153d Mon Sep 17 00:00:00 2001 From: VS Chandra Mourya Date: Fri, 20 Mar 2026 14:57:21 -0700 Subject: [PATCH 07/12] fix(grpc): simplify harmony request typing and final response formatting Signed-off-by: VS Chandra Mourya --- model_gateway/src/routers/grpc/context.rs | 13 +++++ .../grpc/harmony/stages/request_building.rs | 53 ++++--------------- .../harmony/stages/response_processing.rs | 53 ++++--------------- model_gateway/src/routers/grpc/pipeline.rs | 15 +++--- 4 files changed, 40 insertions(+), 94 deletions(-) diff --git a/model_gateway/src/routers/grpc/context.rs b/model_gateway/src/routers/grpc/context.rs index 2c0f559d44..7ce0fad820 100644 --- a/model_gateway/src/routers/grpc/context.rs +++ b/model_gateway/src/routers/grpc/context.rs @@ -72,6 +72,19 @@ impl std::fmt::Display for RequestType { } } +impl std::fmt::Display for FinalResponse { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Chat(_) => write!(f, "Chat"), + Self::Generate(_) => write!(f, "Generate"), + Self::Completion(_) => write!(f, "Completion"), + Self::Embedding(_) => write!(f, "Embedding"), + Self::Classify(_) => write!(f, "Classify"), + Self::Messages(_) => write!(f, "Messages"), + } + } +} + /// Shared components (injected once at creation) pub(crate) struct SharedComponents { pub tokenizer_registry: Arc, diff --git a/model_gateway/src/routers/grpc/harmony/stages/request_building.rs b/model_gateway/src/routers/grpc/harmony/stages/request_building.rs index eeca3cb92e..77952703ed 100644 --- a/model_gateway/src/routers/grpc/harmony/stages/request_building.rs +++ b/model_gateway/src/routers/grpc/harmony/stages/request_building.rs @@ -62,54 +62,19 @@ impl PipelineStage for HarmonyRequestBuildingStage { let request_id = match &ctx.input.request_type { RequestType::Chat(_) => format!("chatcmpl-{}", Uuid::now_v7()), RequestType::Responses(_) => format!("responses-{}", Uuid::now_v7()), - RequestType::Generate(_) => { + request_type @ (RequestType::Generate(_) + | RequestType::Completion(_) + | RequestType::Embedding(_) + | RequestType::Classify(_) + | RequestType::Messages(_)) => { error!( function = "HarmonyRequestBuildingStage::execute", - "Generate request type not supported for Harmony models" + request_type = %request_type, + "{request_type} request type not supported for Harmony models" ); return Err(error::bad_request( - "harmony_generate_not_supported", - "Generate requests are not supported with Harmony models".to_string(), - )); - } - RequestType::Completion(_) => { - error!( - function = "HarmonyRequestBuildingStage::execute", - "Completion request type not supported for Harmony models" - ); - return Err(error::bad_request( - "harmony_completion_not_supported", - "Completion requests are not supported with Harmony models".to_string(), - )); - } - RequestType::Embedding(_) => { - error!( - function = "HarmonyRequestBuildingStage::execute", - "Embedding request type not supported for Harmony models" - ); - return Err(error::bad_request( - "harmony_embedding_not_supported", - "Embedding requests are not supported with Harmony models".to_string(), - )); - } - RequestType::Classify(_) => { - error!( - function = "HarmonyRequestBuildingStage::execute", - "Classify request type not supported for Harmony models" - ); - return Err(error::bad_request( - "harmony_classify_not_supported", - "Classify requests are not supported with Harmony models".to_string(), - )); - } - RequestType::Messages(_) => { - error!( - function = "HarmonyRequestBuildingStage::execute", - "Messages request type not supported for Harmony models" - ); - return Err(error::bad_request( - "harmony_messages_not_supported", - "Messages requests are not supported with Harmony models".to_string(), + "harmony_request_type_not_supported", + format!("{request_type} requests are not supported with Harmony models"), )); } }; diff --git a/model_gateway/src/routers/grpc/harmony/stages/response_processing.rs b/model_gateway/src/routers/grpc/harmony/stages/response_processing.rs index 4bb14ddfde..ce2a6e5469 100644 --- a/model_gateway/src/routers/grpc/harmony/stages/response_processing.rs +++ b/model_gateway/src/routers/grpc/harmony/stages/response_processing.rs @@ -138,54 +138,19 @@ impl PipelineStage for HarmonyResponseProcessingStage { ctx.state.response.responses_iteration_result = Some(iteration_result); Ok(None) } - RequestType::Generate(_) => { + request_type @ (RequestType::Generate(_) + | RequestType::Completion(_) + | RequestType::Embedding(_) + | RequestType::Classify(_) + | RequestType::Messages(_)) => { error!( function = "HarmonyResponseProcessingStage::execute", - "Generate request type not supported in Harmony pipeline" + request_type = %request_type, + "{request_type} request type not supported in Harmony pipeline" ); Err(error::internal_error( - "harmony_generate_not_supported", - "Generate requests not supported in Harmony pipeline", - )) - } - RequestType::Completion(_) => { - error!( - function = "HarmonyResponseProcessingStage::execute", - "Completion request type not supported in Harmony pipeline" - ); - Err(error::internal_error( - "harmony_completion_not_supported", - "Completion requests not supported in Harmony pipeline", - )) - } - RequestType::Embedding(_) => { - error!( - function = "HarmonyResponseProcessingStage::execute", - "Embedding request type not supported in Harmony pipeline" - ); - Err(error::internal_error( - "harmony_embedding_not_supported", - "Embedding requests not supported in Harmony pipeline", - )) - } - RequestType::Classify(_) => { - error!( - function = "HarmonyResponseProcessingStage::execute", - "Classify request type not supported in Harmony pipeline" - ); - Err(error::internal_error( - "harmony_classify_not_supported", - "Classify requests not supported in Harmony pipeline", - )) - } - RequestType::Messages(_) => { - error!( - function = "HarmonyResponseProcessingStage::execute", - "Messages request type not supported in Harmony pipeline" - ); - Err(error::internal_error( - "harmony_messages_not_supported", - "Messages requests not supported in Harmony pipeline", + "harmony_request_type_not_supported", + format!("{request_type} requests not supported in Harmony pipeline"), )) } } diff --git a/model_gateway/src/routers/grpc/pipeline.rs b/model_gateway/src/routers/grpc/pipeline.rs index a1f7ca22a2..3c67e62034 100644 --- a/model_gateway/src/routers/grpc/pipeline.rs +++ b/model_gateway/src/routers/grpc/pipeline.rs @@ -646,14 +646,17 @@ impl RequestPipeline { ); axum::Json(response).into_response() } - Some(FinalResponse::Chat(_)) - | Some(FinalResponse::Generate(_)) - | Some(FinalResponse::Embedding(_)) - | Some(FinalResponse::Classify(_)) - | Some(FinalResponse::Messages(_)) => { + Some( + response_type @ (FinalResponse::Chat(_) + | FinalResponse::Generate(_) + | FinalResponse::Embedding(_) + | FinalResponse::Classify(_) + | FinalResponse::Messages(_)), + ) => { error!( function = "execute_completion", - "Wrong response type: expected Completion" + response_type = %response_type, + "Wrong response type: expected Completion, got {response_type}" ); Metrics::record_router_error( metrics_labels::ROUTER_GRPC, From c6a194eec4c4294c2ddf083e8a384775cdd5eee2 Mon Sep 17 00:00:00 2001 From: VS Chandra Mourya Date: Fri, 20 Mar 2026 15:22:13 -0700 Subject: [PATCH 08/12] minor fixes Signed-off-by: VS Chandra Mourya --- crates/protocols/src/completion.rs | 4 +++- model_gateway/src/routers/grpc/context.rs | 2 +- .../src/routers/grpc/harmony/stages/request_building.rs | 2 +- .../src/routers/grpc/harmony/stages/response_processing.rs | 2 +- 4 files changed, 6 insertions(+), 4 deletions(-) diff --git a/crates/protocols/src/completion.rs b/crates/protocols/src/completion.rs index 01b1adcf9e..f1fbe007a0 100644 --- a/crates/protocols/src/completion.rs +++ b/crates/protocols/src/completion.rs @@ -84,7 +84,9 @@ pub struct CompletionRequest { /// A unique identifier representing your end-user pub user: Option, - // Engine-specific sampling parameters + // ============================================================================= + // Engine-Specific Sampling Parameters + // ============================================================================= /// Top-k sampling parameter (-1 to disable) #[validate(custom(function = "validate_top_k_value"))] pub top_k: Option, diff --git a/model_gateway/src/routers/grpc/context.rs b/model_gateway/src/routers/grpc/context.rs index 7ce0fad820..73dd88f68c 100644 --- a/model_gateway/src/routers/grpc/context.rs +++ b/model_gateway/src/routers/grpc/context.rs @@ -666,7 +666,7 @@ pub(crate) enum ExecutionResult { #[derive(Debug)] #[expect( dead_code, - reason = "Completion final response is introduced in the plumbing PR before later stacked PRs produce it" + reason = "Completion responses are typed in the pipeline before a later stage constructs them" )] pub(crate) enum FinalResponse { Chat(ChatCompletionResponse), diff --git a/model_gateway/src/routers/grpc/harmony/stages/request_building.rs b/model_gateway/src/routers/grpc/harmony/stages/request_building.rs index 77952703ed..571b38869d 100644 --- a/model_gateway/src/routers/grpc/harmony/stages/request_building.rs +++ b/model_gateway/src/routers/grpc/harmony/stages/request_building.rs @@ -73,7 +73,7 @@ impl PipelineStage for HarmonyRequestBuildingStage { "{request_type} request type not supported for Harmony models" ); return Err(error::bad_request( - "harmony_request_type_not_supported", + "not_supported_in_harmony", format!("{request_type} requests are not supported with Harmony models"), )); } diff --git a/model_gateway/src/routers/grpc/harmony/stages/response_processing.rs b/model_gateway/src/routers/grpc/harmony/stages/response_processing.rs index ce2a6e5469..4e574b66a5 100644 --- a/model_gateway/src/routers/grpc/harmony/stages/response_processing.rs +++ b/model_gateway/src/routers/grpc/harmony/stages/response_processing.rs @@ -149,7 +149,7 @@ impl PipelineStage for HarmonyResponseProcessingStage { "{request_type} request type not supported in Harmony pipeline" ); Err(error::internal_error( - "harmony_request_type_not_supported", + "not_supported_in_harmony", format!("{request_type} requests not supported in Harmony pipeline"), )) } From 6e33908e39ef4138d4e4c468d391833d1d99a2a5 Mon Sep 17 00:00:00 2001 From: VS Chandra Mourya Date: Fri, 20 Mar 2026 16:25:31 -0700 Subject: [PATCH 09/12] refactor(grpc): extract wrong-response helpers in pipeline.rs Signed-off-by: VS Chandra Mourya --- model_gateway/src/routers/grpc/pipeline.rs | 242 +++++++++------------ 1 file changed, 108 insertions(+), 134 deletions(-) diff --git a/model_gateway/src/routers/grpc/pipeline.rs b/model_gateway/src/routers/grpc/pipeline.rs index 3c67e62034..4ffe08ba35 100644 --- a/model_gateway/src/routers/grpc/pipeline.rs +++ b/model_gateway/src/routers/grpc/pipeline.rs @@ -61,6 +61,48 @@ pub(crate) struct RequestPipeline { } impl RequestPipeline { + fn wrong_response_type( + &self, + function: &'static str, + expected: &'static str, + response_type: &FinalResponse, + model: &str, + endpoint: &'static str, + ) -> Response { + error!( + function = function, + response_type = %response_type, + "Wrong response type: expected {expected}, got {response_type}" + ); + Metrics::record_router_error( + metrics_labels::ROUTER_GRPC, + self.backend_type, + metrics_labels::CONNECTION_GRPC, + model, + endpoint, + metrics_labels::ERROR_INTERNAL, + ); + error::internal_error("wrong_response_type", "Internal error: wrong response type") + } + + fn no_response_produced( + &self, + function: &'static str, + model: &str, + endpoint: &'static str, + ) -> Response { + error!(function = function, "No response produced by pipeline"); + Metrics::record_router_error( + metrics_labels::ROUTER_GRPC, + self.backend_type, + metrics_labels::CONNECTION_GRPC, + model, + endpoint, + metrics_labels::ERROR_INTERNAL, + ); + error::internal_error("no_response_produced", "No response produced") + } + /// Create a regular (single-worker) pipeline pub fn new_regular( worker_registry: Arc, @@ -430,44 +472,27 @@ impl RequestPipeline { ); axum::Json(response).into_response() } - Some(FinalResponse::Generate(_)) - | Some(FinalResponse::Completion(_)) - | Some(FinalResponse::Embedding(_)) - | Some(FinalResponse::Classify(_)) - | Some(FinalResponse::Messages(_)) => { - error!( - function = "execute_chat", - "Wrong response type: expected Chat, got Generate/Completion/Embedding/Classify/Messages" - ); - Metrics::record_router_error( - metrics_labels::ROUTER_GRPC, - self.backend_type, - metrics_labels::CONNECTION_GRPC, - &request_for_metrics.model, - metrics_labels::ENDPOINT_CHAT, - metrics_labels::ERROR_INTERNAL, - ); - error::internal_error("wrong_response_type", "Internal error: wrong response type") - } - None => { - error!( - function = "execute_chat", - "No response produced by pipeline" - ); - Metrics::record_router_error( - metrics_labels::ROUTER_GRPC, - self.backend_type, - metrics_labels::CONNECTION_GRPC, - &request_for_metrics.model, - metrics_labels::ENDPOINT_CHAT, - metrics_labels::ERROR_INTERNAL, - ); - error::internal_error("no_response_produced", "No response produced") - } + Some( + response_type @ (FinalResponse::Generate(_) + | FinalResponse::Completion(_) + | FinalResponse::Embedding(_) + | FinalResponse::Classify(_) + | FinalResponse::Messages(_)), + ) => self.wrong_response_type( + "execute_chat", + "Chat", + &response_type, + &request_for_metrics.model, + metrics_labels::ENDPOINT_CHAT, + ), + None => self.no_response_produced( + "execute_chat", + &request_for_metrics.model, + metrics_labels::ENDPOINT_CHAT, + ), } } - /// Execute the complete pipeline for a generate request /// Execute the complete pipeline for a generate request pub async fn execute_generate( &self, @@ -536,40 +561,24 @@ impl RequestPipeline { ); axum::Json(response).into_response() } - Some(FinalResponse::Chat(_)) - | Some(FinalResponse::Completion(_)) - | Some(FinalResponse::Embedding(_)) - | Some(FinalResponse::Classify(_)) - | Some(FinalResponse::Messages(_)) => { - error!( - function = "execute_generate", - "Wrong response type: expected Generate, got Chat/Completion/Embedding/Classify/Messages" - ); - Metrics::record_router_error( - metrics_labels::ROUTER_GRPC, - self.backend_type, - metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), - metrics_labels::ENDPOINT_GENERATE, - metrics_labels::ERROR_INTERNAL, - ); - error::internal_error("wrong_response_type", "Internal error: wrong response type") - } - None => { - error!( - function = "execute_generate", - "No response produced by pipeline" - ); - Metrics::record_router_error( - metrics_labels::ROUTER_GRPC, - self.backend_type, - metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), - metrics_labels::ENDPOINT_GENERATE, - metrics_labels::ERROR_INTERNAL, - ); - error::internal_error("no_response_produced", "No response produced") - } + Some( + response_type @ (FinalResponse::Chat(_) + | FinalResponse::Completion(_) + | FinalResponse::Embedding(_) + | FinalResponse::Classify(_) + | FinalResponse::Messages(_)), + ) => self.wrong_response_type( + "execute_generate", + "Generate", + &response_type, + model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + metrics_labels::ENDPOINT_GENERATE, + ), + None => self.no_response_produced( + "execute_generate", + model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + metrics_labels::ENDPOINT_GENERATE, + ), } } @@ -652,37 +661,18 @@ impl RequestPipeline { | FinalResponse::Embedding(_) | FinalResponse::Classify(_) | FinalResponse::Messages(_)), - ) => { - error!( - function = "execute_completion", - response_type = %response_type, - "Wrong response type: expected Completion, got {response_type}" - ); - Metrics::record_router_error( - metrics_labels::ROUTER_GRPC, - self.backend_type, - metrics_labels::CONNECTION_GRPC, - &model, - metrics_labels::ENDPOINT_COMPLETIONS, - metrics_labels::ERROR_INTERNAL, - ); - error::internal_error("wrong_response_type", "Internal error: wrong response type") - } - None => { - error!( - function = "execute_completion", - "No response produced by pipeline" - ); - Metrics::record_router_error( - metrics_labels::ROUTER_GRPC, - self.backend_type, - metrics_labels::CONNECTION_GRPC, - &model, - metrics_labels::ENDPOINT_COMPLETIONS, - metrics_labels::ERROR_INTERNAL, - ); - error::internal_error("no_response_produced", "No response produced") - } + ) => self.wrong_response_type( + "execute_completion", + "Completion", + &response_type, + &model, + metrics_labels::ENDPOINT_COMPLETIONS, + ), + None => self.no_response_produced( + "execute_completion", + &model, + metrics_labels::ENDPOINT_COMPLETIONS, + ), } } @@ -955,40 +945,24 @@ impl RequestPipeline { ); axum::Json(response).into_response() } - Some(FinalResponse::Chat(_)) - | Some(FinalResponse::Generate(_)) - | Some(FinalResponse::Completion(_)) - | Some(FinalResponse::Embedding(_)) - | Some(FinalResponse::Classify(_)) => { - error!( - function = "execute_messages", - "Wrong response type: expected Messages" - ); - Metrics::record_router_error( - metrics_labels::ROUTER_GRPC, - self.backend_type, - metrics_labels::CONNECTION_GRPC, - &request.model, - metrics_labels::ENDPOINT_MESSAGES, - metrics_labels::ERROR_INTERNAL, - ); - error::internal_error("wrong_response_type", "Internal error: wrong response type") - } - None => { - error!( - function = "execute_messages", - "No response produced by pipeline" - ); - Metrics::record_router_error( - metrics_labels::ROUTER_GRPC, - self.backend_type, - metrics_labels::CONNECTION_GRPC, - &request.model, - metrics_labels::ENDPOINT_MESSAGES, - metrics_labels::ERROR_INTERNAL, - ); - error::internal_error("no_response_produced", "No response produced") - } + Some( + response_type @ (FinalResponse::Chat(_) + | FinalResponse::Generate(_) + | FinalResponse::Completion(_) + | FinalResponse::Embedding(_) + | FinalResponse::Classify(_)), + ) => self.wrong_response_type( + "execute_messages", + "Messages", + &response_type, + &request.model, + metrics_labels::ENDPOINT_MESSAGES, + ), + None => self.no_response_produced( + "execute_messages", + &request.model, + metrics_labels::ENDPOINT_MESSAGES, + ), } } From f2018280347031656a9ebc2e38279f4b14e6dd2d Mon Sep 17 00:00:00 2001 From: VS Chandra Mourya Date: Mon, 23 Mar 2026 13:51:38 -0700 Subject: [PATCH 10/12] refactor(grpc): resolve pipeline.rs rebase conflict Signed-off-by: VS Chandra Mourya --- model_gateway/src/routers/grpc/pipeline.rs | 54 +++++++++++----------- 1 file changed, 28 insertions(+), 26 deletions(-) diff --git a/model_gateway/src/routers/grpc/pipeline.rs b/model_gateway/src/routers/grpc/pipeline.rs index 4ffe08ba35..ba10c03e17 100644 --- a/model_gateway/src/routers/grpc/pipeline.rs +++ b/model_gateway/src/routers/grpc/pipeline.rs @@ -502,6 +502,9 @@ impl RequestPipeline { components: Arc, ) -> Response { let start = Instant::now(); + let model = model_id + .clone() + .unwrap_or_else(|| UNKNOWN_MODEL_ID.to_string()); let streaming = request.stream; // Record request start @@ -509,12 +512,12 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_GENERATE, bool_to_static_str(streaming), ); - let mut ctx = RequestContext::for_generate(request, headers, model_id.clone(), components); + let mut ctx = RequestContext::for_generate(request, headers, model_id, components); for stage in self.stages.iter() { match stage.execute(&mut ctx).await { @@ -523,7 +526,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_GENERATE, start.elapsed(), ); @@ -535,7 +538,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_GENERATE, error_type_from_status(response.status()), ); @@ -555,7 +558,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_GENERATE, start.elapsed(), ); @@ -571,12 +574,12 @@ impl RequestPipeline { "execute_generate", "Generate", &response_type, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_GENERATE, ), None => self.no_response_produced( "execute_generate", - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_GENERATE, ), } @@ -684,9 +687,10 @@ impl RequestPipeline { model_id: Option, components: Arc, ) -> Response { + let model = model_id.clone().unwrap_or_else(|| request.model.clone()); debug!( "execute_embeddings: Starting execution for model: {}", - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID) + &model ); let start = Instant::now(); @@ -695,12 +699,12 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_EMBEDDINGS, bool_to_static_str(false), ); - let mut ctx = RequestContext::for_embedding(request, headers, model_id.clone(), components); + let mut ctx = RequestContext::for_embedding(request, headers, model_id, components); for stage in self.stages.iter() { debug!("execute_embeddings: Executing stage: {}", stage.name()); @@ -714,7 +718,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_EMBEDDINGS, start.elapsed(), ); @@ -737,7 +741,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_EMBEDDINGS, error_type_from_status(response.status()), ); @@ -756,7 +760,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_EMBEDDINGS, start.elapsed(), ); @@ -784,10 +788,8 @@ impl RequestPipeline { model_id: Option, components: Arc, ) -> Response { - debug!( - "execute_classify: Starting execution for model: {}", - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID) - ); + let model = model_id.clone().unwrap_or_else(|| request.model.clone()); + debug!("execute_classify: Starting execution for model: {}", &model); let start = Instant::now(); // Record request start @@ -795,12 +797,12 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_CLASSIFY, bool_to_static_str(false), // Classify is never streaming ); - let mut ctx = RequestContext::for_classify(request, headers, model_id.clone(), components); + let mut ctx = RequestContext::for_classify(request, headers, model_id, components); for stage in self.stages.iter() { debug!("execute_classify: Executing stage: {}", stage.name()); @@ -814,7 +816,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_CLASSIFY, start.elapsed(), ); @@ -837,7 +839,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_CLASSIFY, error_type_from_status(response.status()), ); @@ -856,7 +858,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - model_id.as_deref().unwrap_or(UNKNOWN_MODEL_ID), + &model, metrics_labels::ENDPOINT_CLASSIFY, start.elapsed(), ); @@ -1019,7 +1021,7 @@ impl RequestPipeline { | Some(FinalResponse::Messages(_)) => { error!( function = "execute_chat_for_responses", - "Wrong response type: expected Chat, got Generate/Embedding/Classify/Messages" + "Wrong response type: expected Chat, got Generate/Completion/Embedding/Classify/Messages" ); Err(error::internal_error( "wrong_response_type", @@ -1062,8 +1064,8 @@ impl RequestPipeline { // Create RequestContext for this Responses request let mut ctx = RequestContext::for_responses( Arc::new(request.clone()), - None, // No headers needed for internal pipeline execution - None, // Model ID already set in request + None, // No headers needed for internal pipeline execution + Some(request.model.clone()), // Model ID from request harmony_ctx.components.clone(), ); @@ -1126,7 +1128,7 @@ impl RequestPipeline { let mut ctx = RequestContext::for_responses( Arc::new(request.clone()), None, - None, + Some(request.model.clone()), harmony_ctx.components.clone(), ); From bf6aa56425dca892a85af76bbe1c8f8ce6598feb Mon Sep 17 00:00:00 2001 From: Chandra Mourya Date: Mon, 23 Mar 2026 14:12:25 -0700 Subject: [PATCH 11/12] fix(grpc): align PR branch with current model_id handling Signed-off-by: VS Chandra Mourya --- model_gateway/src/routers/grpc/context.rs | 1 + model_gateway/src/routers/grpc/pipeline.rs | 54 +++++++++++----------- 2 files changed, 27 insertions(+), 28 deletions(-) diff --git a/model_gateway/src/routers/grpc/context.rs b/model_gateway/src/routers/grpc/context.rs index b210ebd188..14ac6fae37 100644 --- a/model_gateway/src/routers/grpc/context.rs +++ b/model_gateway/src/routers/grpc/context.rs @@ -277,6 +277,7 @@ impl RequestContext { model_id: Option, components: Arc, ) -> Self { + let model_id = model_id.unwrap_or_else(|| request.model.clone()); Self { input: RequestInput { request_type: RequestType::Completion(request), diff --git a/model_gateway/src/routers/grpc/pipeline.rs b/model_gateway/src/routers/grpc/pipeline.rs index 0328bcdb0c..c76da5fefd 100644 --- a/model_gateway/src/routers/grpc/pipeline.rs +++ b/model_gateway/src/routers/grpc/pipeline.rs @@ -502,9 +502,6 @@ impl RequestPipeline { components: Arc, ) -> Response { let start = Instant::now(); - let model = model_id - .clone() - .unwrap_or_else(|| UNKNOWN_MODEL_ID.to_string()); let streaming = request.stream; // Record request start @@ -512,12 +509,12 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - &model, + &model_id, metrics_labels::ENDPOINT_GENERATE, bool_to_static_str(streaming), ); - let mut ctx = RequestContext::for_generate(request, headers, model_id, components); + let mut ctx = RequestContext::for_generate(request, headers, model_id.clone(), components); for stage in self.stages.iter() { match stage.execute(&mut ctx).await { @@ -526,7 +523,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - &model, + &model_id, metrics_labels::ENDPOINT_GENERATE, start.elapsed(), ); @@ -538,7 +535,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - &model, + &model_id, metrics_labels::ENDPOINT_GENERATE, error_type_from_status(response.status()), ); @@ -558,7 +555,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - &model, + &model_id, metrics_labels::ENDPOINT_GENERATE, start.elapsed(), ); @@ -574,12 +571,12 @@ impl RequestPipeline { "execute_generate", "Generate", &response_type, - &model, + &model_id, metrics_labels::ENDPOINT_GENERATE, ), None => self.no_response_produced( "execute_generate", - &model, + &model_id, metrics_labels::ENDPOINT_GENERATE, ), } @@ -687,10 +684,9 @@ impl RequestPipeline { model_id: String, components: Arc, ) -> Response { - let model = model_id.clone().unwrap_or_else(|| request.model.clone()); debug!( "execute_embeddings: Starting execution for model: {}", - &model + &model_id ); let start = Instant::now(); @@ -699,12 +695,12 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - &model, + &model_id, metrics_labels::ENDPOINT_EMBEDDINGS, bool_to_static_str(false), ); - let mut ctx = RequestContext::for_embedding(request, headers, model_id, components); + let mut ctx = RequestContext::for_embedding(request, headers, model_id.clone(), components); for stage in self.stages.iter() { debug!("execute_embeddings: Executing stage: {}", stage.name()); @@ -718,7 +714,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - &model, + &model_id, metrics_labels::ENDPOINT_EMBEDDINGS, start.elapsed(), ); @@ -741,7 +737,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - &model, + &model_id, metrics_labels::ENDPOINT_EMBEDDINGS, error_type_from_status(response.status()), ); @@ -760,7 +756,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - &model, + &model_id, metrics_labels::ENDPOINT_EMBEDDINGS, start.elapsed(), ); @@ -788,8 +784,10 @@ impl RequestPipeline { model_id: String, components: Arc, ) -> Response { - let model = model_id.clone().unwrap_or_else(|| request.model.clone()); - debug!("execute_classify: Starting execution for model: {}", &model); + debug!( + "execute_classify: Starting execution for model: {}", + &model_id + ); let start = Instant::now(); // Record request start @@ -797,12 +795,12 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - &model, + &model_id, metrics_labels::ENDPOINT_CLASSIFY, bool_to_static_str(false), // Classify is never streaming ); - let mut ctx = RequestContext::for_classify(request, headers, model_id, components); + let mut ctx = RequestContext::for_classify(request, headers, model_id.clone(), components); for stage in self.stages.iter() { debug!("execute_classify: Executing stage: {}", stage.name()); @@ -816,7 +814,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - &model, + &model_id, metrics_labels::ENDPOINT_CLASSIFY, start.elapsed(), ); @@ -839,7 +837,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - &model, + &model_id, metrics_labels::ENDPOINT_CLASSIFY, error_type_from_status(response.status()), ); @@ -858,7 +856,7 @@ impl RequestPipeline { metrics_labels::ROUTER_GRPC, self.backend_type, metrics_labels::CONNECTION_GRPC, - &model, + &model_id, metrics_labels::ENDPOINT_CLASSIFY, start.elapsed(), ); @@ -1021,7 +1019,7 @@ impl RequestPipeline { | Some(FinalResponse::Messages(_)) => { error!( function = "execute_chat_for_responses", - "Wrong response type: expected Chat, got Generate/Completion/Embedding/Classify/Messages" + "Wrong response type: expected Chat, got Generate/Embedding/Classify/Messages" ); Err(error::internal_error( "wrong_response_type", @@ -1064,8 +1062,8 @@ impl RequestPipeline { // Create RequestContext for this Responses request let mut ctx = RequestContext::for_responses( Arc::new(request.clone()), - None, // No headers needed for internal pipeline execution - Some(request.model.clone()), // Model ID from request + None, // No headers needed for internal pipeline execution + request.model.clone(), // Model ID from request harmony_ctx.components.clone(), ); @@ -1128,7 +1126,7 @@ impl RequestPipeline { let mut ctx = RequestContext::for_responses( Arc::new(request.clone()), None, - Some(request.model.clone()), + request.model.clone(), harmony_ctx.components.clone(), ); From a0e43e3d2bf5a2076dfdffcd272872dd0cdd085b Mon Sep 17 00:00:00 2001 From: VS Chandra Mourya Date: Mon, 23 Mar 2026 15:55:18 -0700 Subject: [PATCH 12/12] fix(grpc) fixing coderabbit requests Signed-off-by: VS Chandra Mourya --- model_gateway/src/routers/grpc/context.rs | 3 +-- model_gateway/src/routers/grpc/pipeline.rs | 5 ++--- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/model_gateway/src/routers/grpc/context.rs b/model_gateway/src/routers/grpc/context.rs index 14ac6fae37..3b9dea48ae 100644 --- a/model_gateway/src/routers/grpc/context.rs +++ b/model_gateway/src/routers/grpc/context.rs @@ -274,10 +274,9 @@ impl RequestContext { pub fn for_completion( request: Arc, headers: Option, - model_id: Option, + model_id: String, components: Arc, ) -> Self { - let model_id = model_id.unwrap_or_else(|| request.model.clone()); Self { input: RequestInput { request_type: RequestType::Completion(request), diff --git a/model_gateway/src/routers/grpc/pipeline.rs b/model_gateway/src/routers/grpc/pipeline.rs index c76da5fefd..181c13a984 100644 --- a/model_gateway/src/routers/grpc/pipeline.rs +++ b/model_gateway/src/routers/grpc/pipeline.rs @@ -591,7 +591,7 @@ impl RequestPipeline { &self, request: Arc, headers: Option, - model_id: Option, + _model_id: Option, components: Arc, ) -> Response { let start = Instant::now(); @@ -607,8 +607,7 @@ impl RequestPipeline { bool_to_static_str(streaming), ); - let mut ctx = - RequestContext::for_completion(request, headers, model_id.clone(), components); + let mut ctx = RequestContext::for_completion(request, headers, model.clone(), components); for stage in self.stages.iter() { match stage.execute(&mut ctx).await {