From 95e9ca0e565198efe14a1c3d1f0ec0ea1a793c15 Mon Sep 17 00:00:00 2001 From: yifeng liu <31553858+pallasathena92@users.noreply.github.com> Date: Tue, 11 Aug 2026 08:25:10 -0700 Subject: [PATCH] feat(gateway): honor per-model parser overrides from model cards Parser selection on the gRPC serving path was process-global: one --tool-call-parser / --reasoning-parser name per router, while the tokenizer side is already per-model. A gateway serving two model families could not parse both correctly, and the ModelCard tool_parser/reasoning_parser fields existed with builders but had no read sites. Wire them end to end: - build_model_card maps tool_parser / reasoning_parser worker labels into the card (explicit WorkerSpec cards keep their own values, the same precedence tokenizer_path uses), so overrides flow through the existing label pipeline with no new plumbing. - Worker registration rejects override names the parser registries don't know, instead of shipping silently unparsed output at serve time. - A ParserResolver replaces the configured-name plumbing in the gRPC processors, preparation stages, and shared components: per request, the model's card override wins, then the configured global name, then the factories' existing name-based auto-detection, unchanged. Behavior without overrides is identical; the debug /parse endpoints keep their explicit per-request override. Signed-off-by: yifeng liu <31553858+pallasathena92@users.noreply.github.com> --- model_gateway/src/routers/grpc/context.rs | 8 +- model_gateway/src/routers/grpc/pipeline.rs | 24 +-- .../src/routers/grpc/regular/processor.rs | 52 +++-- .../grpc/regular/stages/chat/preparation.rs | 10 +- .../regular/stages/messages/preparation.rs | 10 +- .../src/routers/grpc/regular/streaming.rs | 57 +++--- model_gateway/src/routers/grpc/router.rs | 8 +- model_gateway/src/routers/grpc/utils/mod.rs | 2 +- .../src/routers/grpc/utils/parsers.rs | 182 +++++++++++++++++- .../src/workflow/steps/local/create_worker.rs | 180 +++++++++++++++++ 10 files changed, 467 insertions(+), 66 deletions(-) diff --git a/model_gateway/src/routers/grpc/context.rs b/model_gateway/src/routers/grpc/context.rs index 618509c7e9..614aaad846 100644 --- a/model_gateway/src/routers/grpc/context.rs +++ b/model_gateway/src/routers/grpc/context.rs @@ -29,6 +29,7 @@ use super::{ EncodeItemBootstrapInfo, ProtoEmbedComplete, ProtoEmbedRequest, ProtoGenerateRequest, ProtoRequest, ProtoStream, }, + utils::ParserResolver, }; use crate::{ middleware::TenantRequestMeta, @@ -144,10 +145,9 @@ pub(crate) struct SharedComponents { pub worker_registry: Arc, pub tool_parser_factory: ToolParserFactory, pub reasoning_parser_factory: ReasoningParserFactory, - /// Configured tool parser name (from CLI `--tool-call-parser`) - pub configured_tool_parser: Option, - /// Configured reasoning parser name (from CLI `--reasoning-parser`) - pub configured_reasoning_parser: Option, + /// Per-request parser-name resolution (model-card override → configured + /// CLI `--tool-call-parser`/`--reasoning-parser` names). + pub parser_resolver: ParserResolver, /// Multimodal processing components (initialized at router creation) pub multimodal: Option>, } diff --git a/model_gateway/src/routers/grpc/pipeline.rs b/model_gateway/src/routers/grpc/pipeline.rs index 741be81bad..1a7c63da26 100644 --- a/model_gateway/src/routers/grpc/pipeline.rs +++ b/model_gateway/src/routers/grpc/pipeline.rs @@ -46,6 +46,7 @@ use super::{ }, streaming, }, + utils, utils::error_type_from_status, }; use crate::{ @@ -138,17 +139,20 @@ impl PipelineDeps { processor::ResponseProcessor, Arc, ) { + let parser_resolver = utils::ParserResolver::new( + self.worker_registry.clone(), + self.configured_tool_parser.clone(), + self.configured_reasoning_parser.clone(), + ); let processor = processor::ResponseProcessor::new( self.tool_parser_factory.clone(), self.reasoning_parser_factory.clone(), - self.configured_tool_parser.clone(), - self.configured_reasoning_parser.clone(), + parser_resolver.clone(), ); let streaming_processor = Arc::new(streaming::StreamingProcessor::new( self.tool_parser_factory.clone(), self.reasoning_parser_factory.clone(), - self.configured_tool_parser.clone(), - self.configured_reasoning_parser.clone(), + parser_resolver, backend, )); (processor, streaming_processor) @@ -165,14 +169,12 @@ impl PipelineDeps { let processor = processor::ResponseProcessor::new( ToolParserFactory::default(), ReasoningParserFactory::default(), - None, - None, + utils::ParserResolver::disabled(), ); let streaming_processor = Arc::new(streaming::StreamingProcessor::new( ToolParserFactory::default(), ReasoningParserFactory::default(), - None, - None, + utils::ParserResolver::disabled(), backend, )); (processor, streaming_processor) @@ -1570,8 +1572,7 @@ mod alias_pipeline_tests { worker_registry, tool_parser_factory: ToolParserFactory::default(), reasoning_parser_factory: ReasoningParserFactory::default(), - configured_tool_parser: None, - configured_reasoning_parser: None, + parser_resolver: utils::ParserResolver::disabled(), multimodal: None, }); let request: GenerateRequest = serde_json::from_value(json!({ @@ -1644,8 +1645,7 @@ mod rate_limit_reserve_tests { worker_registry, tool_parser_factory: ToolParserFactory::default(), reasoning_parser_factory: ReasoningParserFactory::default(), - configured_tool_parser: None, - configured_reasoning_parser: None, + parser_resolver: utils::ParserResolver::disabled(), multimodal: None, }) } diff --git a/model_gateway/src/routers/grpc/regular/processor.rs b/model_gateway/src/routers/grpc/regular/processor.rs index 2275f3eb96..46eccf9251 100644 --- a/model_gateway/src/routers/grpc/regular/processor.rs +++ b/model_gateway/src/routers/grpc/regular/processor.rs @@ -38,22 +38,20 @@ use crate::routers::{ pub(crate) struct ResponseProcessor { pub tool_parser_factory: ToolParserFactory, pub reasoning_parser_factory: ReasoningParserFactory, - pub configured_tool_parser: Option, - pub configured_reasoning_parser: Option, + /// Per-request parser-name resolution (model-card override → configured). + pub parser_resolver: utils::ParserResolver, } impl ResponseProcessor { pub fn new( tool_parser_factory: ToolParserFactory, reasoning_parser_factory: ReasoningParserFactory, - configured_tool_parser: Option, - configured_reasoning_parser: Option, + parser_resolver: utils::ParserResolver, ) -> Self { Self { tool_parser_factory, reasoning_parser_factory, - configured_tool_parser, - configured_reasoning_parser, + parser_resolver, } } @@ -69,6 +67,11 @@ impl ResponseProcessor { history_tool_calls_count: usize, reasoning_parser_available: bool, tool_parser_available: bool, + // Resolved once per request by the caller: keeps every choice of one + // request on the same parser even if the worker registry changes + // between availability check and parsing. + reasoning_parser_name: Option<&str>, + tool_parser_name: Option<&str>, ) -> Result { stop_decoder.reset(); // Decode tokens @@ -109,7 +112,7 @@ impl ResponseProcessor { // across requests, so avoid serializing on the shared pooled mutex. if let Some(mut parser) = utils::create_reasoning_parser( &self.reasoning_parser_factory, - self.configured_reasoning_parser.as_deref(), + reasoning_parser_name, &original_request.model, ) { // If the template injected `` in the prefill (thinking toggle @@ -151,7 +154,7 @@ impl ResponseProcessor { let has_structural_tag = self .tool_parser_factory .registry() - .has_structural_tag_for_parser(self.configured_tool_parser.as_deref()); + .has_structural_tag_for_parser(tool_parser_name); let used_json_schema = if has_structural_tag { false } else { @@ -175,6 +178,7 @@ impl ResponseProcessor { .parse_tool_calls( &processed_text, &original_request.model, + tool_parser_name, original_request.tools.as_deref().unwrap_or(&[]), history_tool_calls_count, ) @@ -241,6 +245,8 @@ impl ResponseProcessor { stop_decoder: &mut StopSequenceDecoder, request_logprobs: bool, ) -> Result { + let reasoning_parser_name = self.parser_resolver.reasoning_parser(&chat_request.model); + let tool_parser_name = self.parser_resolver.tool_parser(&chat_request.model); // Collect all responses from the execution result let all_responses = response_collection::collect_responses(execution_result, request_logprobs).await?; @@ -251,7 +257,7 @@ impl ResponseProcessor { let reasoning_parser_available = chat_request.separate_reasoning && utils::check_reasoning_parser_availability( &self.reasoning_parser_factory, - self.configured_reasoning_parser.as_deref(), + reasoning_parser_name.as_deref(), &chat_request.model, ); @@ -264,7 +270,7 @@ impl ResponseProcessor { && chat_request.tools.is_some() && utils::check_tool_parser_availability( &self.tool_parser_factory, - self.configured_tool_parser.as_deref(), + tool_parser_name.as_deref(), &chat_request.model, ); @@ -296,6 +302,8 @@ impl ResponseProcessor { history_tool_calls_count, reasoning_parser_available, tool_parser_available, + reasoning_parser_name.as_deref(), + tool_parser_name.as_deref(), ) .await { @@ -328,15 +336,14 @@ impl ResponseProcessor { &self, processed_text: &str, model: &str, + // Resolved once per request by the caller (see process_single_choice). + tool_parser_name: Option<&str>, tools: &[Tool], history_tool_calls_count: usize, ) -> (Option>, String) { // Get pooled parser for this model - let pooled_parser = utils::get_tool_parser( - &self.tool_parser_factory, - self.configured_tool_parser.as_deref(), - model, - ); + let pooled_parser = + utils::get_tool_parser(&self.tool_parser_factory, tool_parser_name, model); // Try parsing directly (parser will handle detection internally). Pass the // tool schemas so schema-aware parsers coerce argument types by their @@ -502,6 +509,10 @@ impl ResponseProcessor { tokenizer: Arc, stop_decoder: &mut StopSequenceDecoder, ) -> Result { + let reasoning_parser_name = self + .parser_resolver + .reasoning_parser(&messages_request.model); + let tool_parser_name = self.parser_resolver.tool_parser(&messages_request.model); // Collect all responses (no logprobs for Messages API) let all_responses = response_collection::collect_responses(execution_result, false).await?; @@ -537,7 +548,7 @@ impl ResponseProcessor { // or when the selected parser needs structural special tokens (e.g. Inkling). let reasoning_requires_special_tokens = utils::reasoning_parser_requires_special_tokens( &self.reasoning_parser_factory, - self.configured_reasoning_parser.as_deref(), + reasoning_parser_name.as_deref(), &messages_request.model, ); let separate_reasoning = reasoning_requires_special_tokens @@ -551,7 +562,7 @@ impl ResponseProcessor { let reasoning_parser_available = separate_reasoning && utils::check_reasoning_parser_availability( &self.reasoning_parser_factory, - self.configured_reasoning_parser.as_deref(), + reasoning_parser_name.as_deref(), &messages_request.model, ); @@ -564,7 +575,7 @@ impl ResponseProcessor { && messages_request.tools.is_some() && utils::check_tool_parser_availability( &self.tool_parser_factory, - self.configured_tool_parser.as_deref(), + tool_parser_name.as_deref(), &messages_request.model, ); @@ -624,7 +635,7 @@ impl ResponseProcessor { // across requests, so avoid serializing on the shared pooled mutex. if let Some(mut parser) = utils::create_reasoning_parser( &self.reasoning_parser_factory, - self.configured_reasoning_parser.as_deref(), + reasoning_parser_name.as_deref(), &messages_request.model, ) { // If thinking is effectively ON and template has a toggle, start in reasoning mode. @@ -664,7 +675,7 @@ impl ResponseProcessor { let has_structural_tag = self .tool_parser_factory .registry() - .has_structural_tag_for_parser(self.configured_tool_parser.as_deref()); + .has_structural_tag_for_parser(tool_parser_name.as_deref()); let used_json_schema = !has_structural_tag && matches!( &messages_request.tool_choice, @@ -694,6 +705,7 @@ impl ResponseProcessor { .parse_tool_calls( &processed_text, &messages_request.model, + tool_parser_name.as_deref(), &chat_tools, utils::message_utils::get_history_tool_calls_count_messages( &messages_request, diff --git a/model_gateway/src/routers/grpc/regular/stages/chat/preparation.rs b/model_gateway/src/routers/grpc/regular/stages/chat/preparation.rs index 123038906e..9f6bc06481 100644 --- a/model_gateway/src/routers/grpc/regular/stages/chat/preparation.rs +++ b/model_gateway/src/routers/grpc/regular/stages/chat/preparation.rs @@ -239,7 +239,10 @@ impl ChatPreparationStage { .tool_parser_factory .registry() .generate_tool_constraint( - ctx.components.configured_tool_parser.as_deref(), + ctx.components + .parser_resolver + .tool_parser(&request.model) + .as_deref(), tools, tool_choice, ) @@ -257,7 +260,10 @@ impl ChatPreparationStage { let preserve_reasoning_special_tokens = request.separate_reasoning && utils::reasoning_parser_requires_special_tokens( &ctx.components.reasoning_parser_factory, - ctx.components.configured_reasoning_parser.as_deref(), + ctx.components + .parser_resolver + .reasoning_parser(&request.model) + .as_deref(), &request.model, ); diff --git a/model_gateway/src/routers/grpc/regular/stages/messages/preparation.rs b/model_gateway/src/routers/grpc/regular/stages/messages/preparation.rs index 39acddb6cd..701d544b4b 100644 --- a/model_gateway/src/routers/grpc/regular/stages/messages/preparation.rs +++ b/model_gateway/src/routers/grpc/regular/stages/messages/preparation.rs @@ -267,7 +267,10 @@ impl MessagePreparationStage { .tool_parser_factory .registry() .generate_tool_constraint( - ctx.components.configured_tool_parser.as_deref(), + ctx.components + .parser_resolver + .tool_parser(&request.model) + .as_deref(), &filtered_tools, tool_choice, ) @@ -290,7 +293,10 @@ impl MessagePreparationStage { let preserve_reasoning_special_tokens = utils::reasoning_parser_requires_special_tokens( &ctx.components.reasoning_parser_factory, - ctx.components.configured_reasoning_parser.as_deref(), + ctx.components + .parser_resolver + .reasoning_parser(&request.model) + .as_deref(), &request.model, ); diff --git a/model_gateway/src/routers/grpc/regular/streaming.rs b/model_gateway/src/routers/grpc/regular/streaming.rs index 473af51a1e..3e39e234a9 100644 --- a/model_gateway/src/routers/grpc/regular/streaming.rs +++ b/model_gateway/src/routers/grpc/regular/streaming.rs @@ -79,8 +79,8 @@ struct CompletionStreamOutcome { pub(crate) struct StreamingProcessor { tool_parser_factory: ToolParserFactory, reasoning_parser_factory: ReasoningParserFactory, - configured_tool_parser: Option, - configured_reasoning_parser: Option, + /// Per-request parser-name resolution (model-card override → configured). + parser_resolver: utils::ParserResolver, backend_type: &'static str, } @@ -101,15 +101,13 @@ impl StreamingProcessor { pub fn new( tool_parser_factory: ToolParserFactory, reasoning_parser_factory: ReasoningParserFactory, - configured_tool_parser: Option, - configured_reasoning_parser: Option, + parser_resolver: utils::ParserResolver, backend_type: &'static str, ) -> Self { Self { tool_parser_factory, reasoning_parser_factory, - configured_tool_parser, - configured_reasoning_parser, + parser_resolver, backend_type, } } @@ -319,11 +317,15 @@ impl StreamingProcessor { let created = dispatch.created; let system_fingerprint = dispatch.weight_version.as_deref(); + // Per-request effective parser names (model-card override → configured). + let reasoning_parser_name = self.parser_resolver.reasoning_parser(model); + let tool_parser_name = self.parser_resolver.tool_parser(model); + // Check parser availability once upfront (log warning only once per request) let reasoning_parser_available = separate_reasoning && utils::check_reasoning_parser_availability( &self.reasoning_parser_factory, - self.configured_reasoning_parser.as_deref(), + reasoning_parser_name.as_deref(), model, ); @@ -344,7 +346,7 @@ impl StreamingProcessor { let has_structural_tag = self .tool_parser_factory .registry() - .has_structural_tag_for_parser(self.configured_tool_parser.as_deref()); + .has_structural_tag_for_parser(tool_parser_name.as_deref()); let used_json_schema = if has_structural_tag { false } else { @@ -365,7 +367,7 @@ impl StreamingProcessor { let tool_parser_available = tools.is_some() && utils::check_tool_parser_availability( &self.tool_parser_factory, - self.configured_tool_parser.as_deref(), + tool_parser_name.as_deref(), model, ); @@ -490,6 +492,7 @@ impl StreamingProcessor { &mut reasoning_parsers, thinking_override, think_in_prefill, + reasoning_parser_name.as_deref(), request_id, model, created, @@ -537,6 +540,7 @@ impl StreamingProcessor { &mut tool_parsers, &mut has_tool_calls, tools_ref, + tool_parser_name.as_deref(), request_id, model, created, @@ -1357,6 +1361,10 @@ impl StreamingProcessor { reasoning_parsers: &mut HashMap>>>, thinking_override: bool, think_in_prefill: bool, + // Resolved once per request by the caller: re-resolving here could + // disagree with the upfront availability check if the worker registry + // changed mid-stream, turning the `expect` below into a panic. + reasoning_parser_name: Option<&str>, request_id: &str, model: &str, created: u64, @@ -1370,7 +1378,7 @@ impl StreamingProcessor { reasoning_parsers.entry(index).or_insert_with(|| { let mut parser = utils::create_reasoning_parser( &self.reasoning_parser_factory, - self.configured_reasoning_parser.as_deref(), + reasoning_parser_name, model, ) .expect("Parser should be available - checked upfront"); @@ -1480,6 +1488,8 @@ impl StreamingProcessor { tool_parsers: &mut HashMap>>>, has_tool_calls: &mut HashMap, tools: &[Tool], + // Resolved once per request by the caller (see process_reasoning_stream). + tool_parser_name: Option<&str>, request_id: &str, model: &str, created: u64, @@ -1499,12 +1509,8 @@ impl StreamingProcessor { utils::create_tool_parser(&self.tool_parser_factory, Some("json"), model) .expect("JSON parser should be available") } else { - utils::create_tool_parser( - &self.tool_parser_factory, - self.configured_tool_parser.as_deref(), - model, - ) - .expect("Parser should be available - checked upfront") + utils::create_tool_parser(&self.tool_parser_factory, tool_parser_name, model) + .expect("Parser should be available - checked upfront") }; Arc::new(tokio::sync::Mutex::new(parser)) }); @@ -1653,13 +1659,15 @@ impl StreamingProcessor { reasoning_parser: &mut Option>>>, thinking_override: bool, think_in_prefill: bool, + // Resolved once per request by the caller (see process_reasoning_stream). + reasoning_parser_name: Option<&str>, model: &str, ) -> (String, String, bool) { // Lazily create parser if reasoning_parser.is_none() { if let Some(mut parser) = utils::create_reasoning_parser( &self.reasoning_parser_factory, - self.configured_reasoning_parser.as_deref(), + reasoning_parser_name, model, ) { if thinking_override { @@ -1884,11 +1892,15 @@ impl StreamingProcessor { // engine output (the backend has no stop-string detection over ZMQ). let mut stopped = false; + // Per-request effective parser names (model-card override → configured). + let reasoning_parser_name = self.parser_resolver.reasoning_parser(model); + let tool_parser_name = self.parser_resolver.tool_parser(model); + // Check parser availability once upfront. Run parser when the user explicitly // enabled thinking, or when the selected parser needs structural special tokens. let reasoning_requires_special_tokens = utils::reasoning_parser_requires_special_tokens( &self.reasoning_parser_factory, - self.configured_reasoning_parser.as_deref(), + reasoning_parser_name.as_deref(), model, ); let separate_reasoning = reasoning_requires_special_tokens @@ -1902,7 +1914,7 @@ impl StreamingProcessor { let reasoning_parser_available = separate_reasoning && utils::check_reasoning_parser_availability( &self.reasoning_parser_factory, - self.configured_reasoning_parser.as_deref(), + reasoning_parser_name.as_deref(), model, ); @@ -1928,14 +1940,14 @@ impl StreamingProcessor { && tool_choice_enabled && utils::check_tool_parser_availability( &self.tool_parser_factory, - self.configured_tool_parser.as_deref(), + tool_parser_name.as_deref(), model, ); let has_structural_tag = self .tool_parser_factory .registry() - .has_structural_tag_for_parser(self.configured_tool_parser.as_deref()); + .has_structural_tag_for_parser(tool_parser_name.as_deref()); let used_json_schema = !has_structural_tag && matches!( &original_request.tool_choice, @@ -1966,7 +1978,7 @@ impl StreamingProcessor { let parser_name = if used_json_schema { Some("json") } else { - self.configured_tool_parser.as_deref() + tool_parser_name.as_deref() }; utils::create_tool_parser(&self.tool_parser_factory, parser_name, model) } else { @@ -2045,6 +2057,7 @@ impl StreamingProcessor { &mut reasoning_parser, thinking_override, think_in_prefill, + reasoning_parser_name.as_deref(), model, ) .await diff --git a/model_gateway/src/routers/grpc/router.rs b/model_gateway/src/routers/grpc/router.rs index e54eb59b19..0591209243 100644 --- a/model_gateway/src/routers/grpc/router.rs +++ b/model_gateway/src/routers/grpc/router.rs @@ -33,6 +33,7 @@ use super::{ multimodal::MultimodalComponents, pipeline::{Endpoint, PipelineDeps, RequestPipeline}, regular::responses, + utils::ParserResolver, }; use crate::{ app_context::AppContext, @@ -354,8 +355,11 @@ impl GrpcRouter { worker_registry: worker_registry.clone(), tool_parser_factory: tool_parser_factory.clone(), reasoning_parser_factory: reasoning_parser_factory.clone(), - configured_tool_parser: ctx.configured_tool_parser.clone(), - configured_reasoning_parser: ctx.configured_reasoning_parser.clone(), + parser_resolver: ParserResolver::new( + worker_registry.clone(), + ctx.configured_tool_parser.clone(), + ctx.configured_reasoning_parser.clone(), + ), multimodal, }); diff --git a/model_gateway/src/routers/grpc/utils/mod.rs b/model_gateway/src/routers/grpc/utils/mod.rs index 5199bd0953..517e25fece 100644 --- a/model_gateway/src/routers/grpc/utils/mod.rs +++ b/model_gateway/src/routers/grpc/utils/mod.rs @@ -22,7 +22,7 @@ pub(crate) use logprobs::{ pub(crate) use metrics::{error_type_from_status, route_to_endpoint}; pub(crate) use parsers::{ check_reasoning_parser_availability, check_tool_parser_availability, create_reasoning_parser, - create_tool_parser, get_tool_parser, reasoning_parser_requires_special_tokens, + create_tool_parser, get_tool_parser, reasoning_parser_requires_special_tokens, ParserResolver, }; // `pub` (not `pub(crate)`) so the Go bindings can reuse the gateway's reasoning // detection instead of duplicating it. diff --git a/model_gateway/src/routers/grpc/utils/parsers.rs b/model_gateway/src/routers/grpc/utils/parsers.rs index 3d47966211..51ea413ed8 100644 --- a/model_gateway/src/routers/grpc/utils/parsers.rs +++ b/model_gateway/src/routers/grpc/utils/parsers.rs @@ -1,10 +1,12 @@ //! Reasoning and tool parser helpers. +use std::sync::Arc; + use llm_tokenizer::{ chat_template::{ThinkingKeyName, ThinkingToggle}, traits::Tokenizer, }; -use openai_protocol::chat::thinking_from_reasoning_effort; +use openai_protocol::{chat::thinking_from_reasoning_effort, model_card::ModelCard}; use reasoning_parser::{ParserFactory as ReasoningParserFactory, ReasoningParser}; use serde_json::Value; use tool_parser::{ @@ -12,6 +14,101 @@ use tool_parser::{ }; use tracing::warn; +use crate::worker::WorkerRegistry; + +/// Per-request parser-name resolution. +/// +/// Precedence: the model's `ModelCard` override (`tool_parser` / +/// `reasoning_parser`, populated from worker labels or an explicit +/// `WorkerSpec` card) → the process-wide configured name +/// (`--tool-call-parser` / `--reasoning-parser`) → `None`, which lets the +/// factory helpers fall back to their name-based auto-detection, unchanged. +/// +/// Lookups borrow straight from worker metadata (no card clones); only the +/// resolved name is cloned. +#[derive(Clone)] +pub(crate) struct ParserResolver { + /// `None` disables card lookups (parser-free endpoints, tests). + worker_registry: Option>, + configured_tool_parser: Option, + configured_reasoning_parser: Option, +} + +impl ParserResolver { + pub(crate) fn new( + worker_registry: Arc, + configured_tool_parser: Option, + configured_reasoning_parser: Option, + ) -> Self { + Self { + worker_registry: Some(worker_registry), + configured_tool_parser, + configured_reasoning_parser, + } + } + + /// Resolver that never consults model cards and carries no configured + /// names — preserves the parser-free endpoints' behavior. + pub(crate) fn disabled() -> Self { + Self { + worker_registry: None, + configured_tool_parser: None, + configured_reasoning_parser: None, + } + } + + /// Effective tool-parser name for `model`, if any. + pub(crate) fn tool_parser(&self, model: &str) -> Option { + self.card_parser(model, |card| card.tool_parser.as_ref()) + .or_else(|| self.configured_tool_parser.clone()) + } + + /// Effective reasoning-parser name for `model`, if any. + pub(crate) fn reasoning_parser(&self, model: &str) -> Option { + self.card_parser(model, |card| card.reasoning_parser.as_ref()) + .or_else(|| self.configured_reasoning_parser.clone()) + } + + fn card_parser( + &self, + model: &str, + pick: impl Fn(&ModelCard) -> Option<&String>, + ) -> Option { + let registry = self.worker_registry.as_ref()?; + // Cards built by the label pipeline agree across workers of one model; + // if they don't (mixed labels, e.g. mid rolling-upgrade), pick the + // lexicographically smallest so resolution is deterministic rather + // than registry-iteration-order dependent. Registration logs a + // warning for the conflict; here it's debug (per-request hot path). + let mut chosen: Option = None; + let mut conflict = false; + for worker in registry.get_by_model(model).iter() { + let Some(name) = worker.metadata().spec.models.find(model).and_then(&pick) else { + continue; + }; + match &chosen { + None => chosen = Some(name.clone()), + Some(existing) if existing != name => { + conflict = true; + if name < existing { + chosen = Some(name.clone()); + } + } + Some(_) => {} + } + } + if conflict { + tracing::debug!( + model, + chosen = chosen.as_deref(), + "Workers for this model declare conflicting parser overrides; \ + using the lexicographically smallest" + ); + } + chosen + } +} + /// Determine if thinking is effectively ON based on the template's thinking /// toggle and the user's request. /// @@ -354,3 +451,86 @@ mod tests { )); } } + +#[cfg(test)] +mod parser_resolver_tests { + use super::*; + use crate::worker::{BasicWorkerBuilder, WorkerRegistry, WorkerType}; + + fn registry_with_card(card: ModelCard) -> Arc { + let registry = Arc::new(WorkerRegistry::new()); + let worker = BasicWorkerBuilder::new("http://w1:8000") + .model(card) + .worker_type(WorkerType::Regular) + .build(); + registry.register(Arc::new(worker)); + registry + } + + #[test] + fn card_override_wins_over_configured() { + let registry = registry_with_card( + ModelCard::new("m") + .with_tool_parser("json") + .with_reasoning_parser("basic"), + ); + let resolver = ParserResolver::new( + registry, + Some("mistral".to_string()), + Some("deepseek_r1".to_string()), + ); + assert_eq!(resolver.tool_parser("m").as_deref(), Some("json")); + assert_eq!(resolver.reasoning_parser("m").as_deref(), Some("basic")); + } + + #[test] + fn falls_back_to_configured_without_card_override() { + let registry = registry_with_card(ModelCard::new("m")); + let resolver = ParserResolver::new( + registry, + Some("mistral".to_string()), + Some("deepseek_r1".to_string()), + ); + assert_eq!(resolver.tool_parser("m").as_deref(), Some("mistral")); + assert_eq!( + resolver.reasoning_parser("m").as_deref(), + Some("deepseek_r1") + ); + // Unknown model: no card, same configured fallback. + assert_eq!(resolver.tool_parser("other").as_deref(), Some("mistral")); + } + + #[test] + fn no_override_and_no_configured_resolves_none() { + let registry = registry_with_card(ModelCard::new("m")); + let resolver = ParserResolver::new(registry, None, None); + assert_eq!(resolver.tool_parser("m"), None); + assert_eq!(resolver.reasoning_parser("m"), None); + } + + #[test] + fn disabled_resolver_never_resolves() { + let resolver = ParserResolver::disabled(); + assert_eq!(resolver.tool_parser("m"), None); + assert_eq!(resolver.reasoning_parser("m"), None); + } + + #[test] + fn conflicting_overrides_resolve_deterministically() { + // Two same-model workers with different overrides: resolution must + // not depend on registration/iteration order — the lexicographically + // smallest name wins either way. + for (first, second) in [("zebra", "alpha"), ("alpha", "zebra")] { + let registry = Arc::new(WorkerRegistry::new()); + for (i, name) in [first, second].iter().enumerate() { + let worker = BasicWorkerBuilder::new(format!("http://w{i}:8000")) + .model(ModelCard::new("m").with_tool_parser(*name)) + .worker_type(WorkerType::Regular) + .build(); + registry.register(Arc::new(worker)); + } + let resolver = ParserResolver::new(registry, None, None); + assert_eq!(resolver.tool_parser("m").as_deref(), Some("alpha")); + } + } +} diff --git a/model_gateway/src/workflow/steps/local/create_worker.rs b/model_gateway/src/workflow/steps/local/create_worker.rs index b5ffc63ec1..e237fc805c 100644 --- a/model_gateway/src/workflow/steps/local/create_worker.rs +++ b/model_gateway/src/workflow/steps/local/create_worker.rs @@ -92,6 +92,30 @@ impl StepExecutor for CreateLocalWorkerStep { &app_context.router_config.model_aliases, ); + // A parser override naming an unknown parser would silently ship + // unparsed output at serve time — fail registration loudly instead + // (mirrors the fail-fast AppContext applies to the global CLI names). + validate_parser_overrides( + &model_card, + &config.url, + app_context.tool_parser_factory.as_ref(), + app_context.reasoning_parser_factory.as_ref(), + ) + .map_err(|message| WorkflowError::StepFailed { + step_id: StepId::new("create_worker"), + message, + })?; + + // Mixed overrides across same-model workers are a misconfiguration + // (except transiently during rolling upgrades): resolution picks one + // deterministically, but only one family parses correctly. Warn, don't + // fail — failing would block rolling upgrades that change the parser. + warn_on_conflicting_parser_overrides( + &model_card, + &config.url, + &app_context.worker_registry, + ); + let runtime_type = match context.data.detected_runtime_type.as_deref() { Some(s) => s.parse::().unwrap_or(config.runtime_type), None => config.runtime_type, @@ -263,6 +287,75 @@ fn resolve_model_id<'a>(config: &'a WorkerSpec, labels: &'a HashMap, + reasoning_parser_factory: Option<&reasoning_parser::ParserFactory>, +) -> Result<(), String> { + if let (Some(name), Some(factory)) = (card.tool_parser.as_deref(), tool_parser_factory) { + if !factory.registry().has_parser(name) { + return Err(format!( + "worker {} declares unknown tool_parser '{}' for model '{}'", + worker_url, name, card.id + )); + } + } + if let (Some(name), Some(factory)) = + (card.reasoning_parser.as_deref(), reasoning_parser_factory) + { + if !factory.registry().has_parser(name) { + return Err(format!( + "worker {} declares unknown reasoning_parser '{}' for model '{}'", + worker_url, name, card.id + )); + } + } + Ok(()) +} + fn build_model_card( model_id: &str, config: &WorkerSpec, @@ -328,6 +421,21 @@ fn build_model_card( .cloned(); } + // Per-model parser overrides: `tool_parser` / `reasoning_parser` labels + // (from backend metadata or WorkerSpec.labels) pin the parser for this + // model, overriding the process-wide `--tool-call-parser` / + // `--reasoning-parser` names on the gRPC serving path. An explicit card + // keeps its own values (same precedence as tokenizer_path above). + if card.tool_parser.is_none() { + card.tool_parser = labels.get("tool_parser").filter(|s| !s.is_empty()).cloned(); + } + if card.reasoning_parser.is_none() { + card.reasoning_parser = labels + .get("reasoning_parser") + .filter(|s| !s.is_empty()) + .cloned(); + } + // Infer model_type capabilities from discovered signals let has_vision = labels .get("supports_vision") @@ -608,4 +716,76 @@ mod tests { assert!(card.aliases.contains(&"glm-5.2".to_string())); assert_eq!(card.aliases.len(), 2); } + + #[test] + fn parser_override_labels_flow_into_card() { + let spec = WorkerSpec::new("http://worker:8080"); + let labels = HashMap::from([ + ("tool_parser".to_string(), "json".to_string()), + ("reasoning_parser".to_string(), "basic".to_string()), + ("model_type".to_string(), "llama".to_string()), + ]); + + let card = build_model_card("m", &spec, &labels, &HashMap::new()); + assert_eq!(card.tool_parser.as_deref(), Some("json")); + assert_eq!(card.reasoning_parser.as_deref(), Some("basic")); + + // Empty label values are treated as unset. + let labels = HashMap::from([("tool_parser".to_string(), String::new())]); + let card = build_model_card("m", &spec, &labels, &HashMap::new()); + assert_eq!(card.tool_parser, None); + } + + #[test] + fn explicit_card_parser_wins_over_labels() { + // A user-provided WorkerSpec card keeps its parser fields; labels + // only fill gaps (same precedence as tokenizer_path). + let mut spec = WorkerSpec::new("http://worker:8080"); + spec.models = openai_protocol::worker::WorkerModels::Single(Box::new( + ModelCard::new("m").with_tool_parser("pythonic"), + )); + let labels = HashMap::from([ + ("tool_parser".to_string(), "json".to_string()), + ("reasoning_parser".to_string(), "basic".to_string()), + ]); + + let card = build_model_card("m", &spec, &labels, &HashMap::new()); + assert_eq!(card.tool_parser.as_deref(), Some("pythonic")); + // The explicit card left reasoning unset — the label fills it. + assert_eq!(card.reasoning_parser.as_deref(), Some("basic")); + } + + #[test] + fn unknown_parser_override_fails_validation() { + let tool_factory = tool_parser::ParserFactory::default(); + let reasoning_factory = reasoning_parser::ParserFactory::default(); + + // No overrides → nothing to validate. + let card = ModelCard::new("m"); + assert!(validate_parser_overrides( + &card, + "http://w:1", + Some(&tool_factory), + Some(&reasoning_factory) + ) + .is_ok()); + + // Known tool parser passes; unknown names fail loudly. + let card = ModelCard::new("m").with_tool_parser("json"); + assert!(validate_parser_overrides(&card, "http://w:1", Some(&tool_factory), None).is_ok()); + + let card = ModelCard::new("m").with_tool_parser("definitely-not-a-parser"); + let err = validate_parser_overrides(&card, "http://w:1", Some(&tool_factory), None) + .expect_err("unknown tool parser must fail registration"); + assert!(err.contains("definitely-not-a-parser"), "{err}"); + + let card = ModelCard::new("m").with_reasoning_parser("definitely-not-a-parser"); + let err = validate_parser_overrides(&card, "http://w:1", None, Some(&reasoning_factory)) + .expect_err("unknown reasoning parser must fail registration"); + assert!(err.contains("definitely-not-a-parser"), "{err}"); + + // Absent factories skip validation (parsers unused in that config). + let card = ModelCard::new("m").with_tool_parser("definitely-not-a-parser"); + assert!(validate_parser_overrides(&card, "http://w:1", None, None).is_ok()); + } }