refactor(zmq): clients finalize their own requests (no more stage-level transport probes)

model_gateway/src/routers/grpc/backend_client.rs

@@ -22,11 +22,12 @@ use crate::{
         client::{
             GenerateRequestBuildOptions, GrpcClient, HealthCheckResponse, ModelInfo, ServerInfo,
         },
+        common::stages::helpers,
         proto_wrapper::{
             finish_tokenspeed_request, finish_vllm_request, ProtoEmbedComplete, ProtoEmbedRequest,
             ProtoGenerateRequest, ProtoStream,
         },
-        zmq_client::ZmqEngineClient,
+        zmq_client::{fold_tokenizer_eos_backstop, ZmqEngineClient},
         MultimodalData,
     },
     worker::RuntimeType,
@@ -55,6 +56,32 @@ impl BackendClient {
         matches!(self, Self::Zmq(_))
     }
 
+    /// Finalize a built generate request for this backend's wire: resolve
+    /// string `stop`s the engine cannot match (token-only wires and SGLang's
+    /// `skip_tokenizer_init` workers) into `stop_token_ids`, folding in EOS
+    /// where the frontend owns stopping.
+    ///
+    /// Returns the router's residual obligation: the stop strings the engine
+    /// will never see, which response processing must trim from output text.
+    /// Empty when the engine matches stops server-side. This is the client's
+    /// own policy — callers need no transport knowledge.
+    pub fn finalize_generate_request(
+        &self,
+        request: &mut ProtoGenerateRequest,
+        tokenizer: Option<&std::sync::Arc<dyn llm_tokenizer::traits::Tokenizer>>,
+    ) -> Vec<String> {
+        let token_only_wire = self.is_zmq();
+        let router_stops = helpers::resolve_string_stops(request, tokenizer, token_only_wire);
+        if let Self::Zmq(client) = self {
+            // EngineCore has no tokenizer, so stopping at EOS is this
+            // frontend's job; TokenSpeed's scheduler stops at EOS itself.
+            if client.runtime() != RuntimeType::TokenSpeed {
+                fold_tokenizer_eos_backstop(request, tokenizer);
+            }
+        }
+        router_stops
+    }
+
     /// Local liveness. gRPC has no cheap local flag (it uses a health RPC), so
     /// this reports `true` for gRPC; ZMQ reflects its connection liveness.
     pub fn is_alive(&self) -> bool {
@@ -218,7 +245,7 @@ impl BackendClient {
                         )
                     })
                 }
-                _ => {
+                RuntimeType::Vllm => {
                     let vllm_mm = zmq_vllm_mm(options.multimodal_inputs)?;
                     finish_vllm_request(vllm_mm, |mm| {
                         VllmEngineClient::build_generate_request_from_chat(
@@ -231,6 +258,9 @@ impl BackendClient {
                         )
                     })
                 }
+                other => Err(format!(
+                    "ZMQ backend reports unsupported runtime {other:?}; expected vLLM or TokenSpeed"
+                )),
             },
         }
     }
@@ -263,7 +293,7 @@ impl BackendClient {
                         )
                     })
                 }
-                _ => {
+                RuntimeType::Vllm => {
                     let vllm_mm = zmq_vllm_mm(options.multimodal_inputs)?;
                     finish_vllm_request(vllm_mm, |mm| {
                         VllmEngineClient::build_generate_request_from_messages(
@@ -276,6 +306,9 @@ impl BackendClient {
                         )
                     })
                 }
+                other => Err(format!(
+                    "ZMQ backend reports unsupported runtime {other:?}; expected vLLM or TokenSpeed"
+                )),
             },
         }
     }
@@ -301,7 +334,7 @@ impl BackendClient {
                     )?;
                     Ok(ProtoGenerateRequest::TokenSpeed(Box::new(req)))
                 }
-                _ => {
+                RuntimeType::Vllm => {
                     let req = VllmEngineClient::build_generate_request_from_completion(
                         request_id,
                         body,
@@ -310,6 +343,9 @@ impl BackendClient {
                     )?;
                     Ok(ProtoGenerateRequest::Vllm(Box::new(req)))
                 }
+                other => Err(format!(
+                    "ZMQ backend reports unsupported runtime {other:?}; expected vLLM or TokenSpeed"
+                )),
             },
         }
     }
@@ -335,7 +371,7 @@ impl BackendClient {
                     )?;
                     Ok(ProtoGenerateRequest::TokenSpeed(Box::new(req)))
                 }
-                _ => {
+                RuntimeType::Vllm => {
                     let req = VllmEngineClient::build_plain_generate_request(
                         request_id,
                         body,
@@ -344,6 +380,9 @@ impl BackendClient {
                     )?;
                     Ok(ProtoGenerateRequest::Vllm(Box::new(req)))
                 }
+                other => Err(format!(
+                    "ZMQ backend reports unsupported runtime {other:?}; expected vLLM or TokenSpeed"
+                )),
             },
         }
     }

model_gateway/src/routers/grpc/common/stages/helpers.rs

@@ -364,43 +364,36 @@ fn encode_single_token_stops(
 /// any single-token stop as a `stop_token_ids` entry for early stopping; the
 /// router-side decoder handles the rest. This is the single resolution point
 /// shared by SGLang gRPC and every ZMQ backend.
+///
+/// Returns the stop strings that were stripped — the router's residual
+/// obligation: the engine will never match these, so response processing must
+/// trim them from the output text. Empty when the engine matches server-side.
 pub(crate) fn resolve_string_stops(
     request: &mut ProtoGenerateRequest,
     tokenizer: Option<&Arc<dyn Tokenizer>>,
-    is_zmq: bool,
-) {
+    token_only_wire: bool,
+) -> Vec<String> {
     // SGLang always needs it; the vLLM and TokenSpeed protos only when talking
-    // to a ZMQ backend (which is the sole path either reaches token-only).
+    // to a token-only wire (direct-ZMQ, the sole path either reaches that way).
     match request {
         ProtoGenerateRequest::Sglang(req) => {
             if let Some(params) = req.sampling_params.as_mut() {
                 let stops = std::mem::take(&mut params.stop);
-                encode_single_token_stops(stops, &mut params.stop_token_ids, tokenizer);
+                encode_single_token_stops(stops.clone(), &mut params.stop_token_ids, tokenizer);
+                return stops;
             }
         }
-        ProtoGenerateRequest::Vllm(req) if is_zmq => {
+        ProtoGenerateRequest::Vllm(req) if token_only_wire => {
+            // EOS injection for the tokenizer-less EngineCore is the ZMQ
+            // client's own policy (zmq_client::fold_tokenizer_eos_backstop),
+            // not part of shared stop resolution.
             if let Some(params) = req.sampling_params.as_mut() {
                 let stops = std::mem::take(&mut params.stop);
-                encode_single_token_stops(stops, &mut params.stop_token_ids, tokenizer);
-                // The engine only stops at EOS when the frontend supplies the
-                // ids, and the connect-time model-dir resolution has nothing
-                // to read when the worker's model id is a repo id rather than
-                // a local path. The tokenizer carries the merged EOS set, so
-                // fold it into the stop tokens as the always-available
-                // backstop — without it an uncapped request generates to the
-                // full context window.
-                if !params.ignore_eos {
-                    if let Some(tokenizer) = tokenizer {
-                        for &id in tokenizer.eos_token_ids() {
-                            if !params.stop_token_ids.contains(&id) {
-                                params.stop_token_ids.push(id);
-                            }
-                        }
-                    }
-                }
+                encode_single_token_stops(stops.clone(), &mut params.stop_token_ids, tokenizer);
+                return stops;
             }
         }
-        ProtoGenerateRequest::TokenSpeed(req) if is_zmq => {
+        ProtoGenerateRequest::TokenSpeed(req) if token_only_wire => {
             // TokenSpeed over ZMQ receives token ids only, so its wire
             // translation drops raw `stop` strings; without this a single-token
             // user stop would never reach the engine as a `stop_token_ids`
@@ -409,11 +402,13 @@ pub(crate) fn resolve_string_stops(
             // stops at EOS itself, so its translation carries no frontend ids.
             if let Some(params) = req.sampling_params.as_mut() {
                 let stops = std::mem::take(&mut params.stop);
-                encode_single_token_stops(stops, &mut params.stop_token_ids, tokenizer);
+                encode_single_token_stops(stops.clone(), &mut params.stop_token_ids, tokenizer);
+                return stops;
             }
         }
         _ => {}
     }
+    Vec::new()
 }
 
 /// Inject PD bootstrap metadata for SGLang if needed.
@@ -698,39 +693,14 @@ mod stop_resolution_tests {
         );
         assert!(params.stop_token_ids.is_empty());
 
-        // ZMQ vLLM (EngineCore sees token ids only) resolves like SGLang, and
-        // gains the tokenizer's EOS ids so generation always terminates.
+        // ZMQ vLLM (EngineCore sees token ids only) resolves like SGLang.
+        // EOS injection is the ZMQ client's own step, not stop resolution's
+        // (see zmq_client::fold_tokenizer_eos_backstop tests).
         let mut zmq = vllm_request(vec!["."], vec![]);
         resolve_string_stops(&mut zmq, Some(&mock_tokenizer()), true);
         let params = vllm_params(&zmq);
         assert!(params.stop.is_empty(), "ZMQ vLLM stop cleared");
-        assert_eq!(params.stop_token_ids, vec![6, 999]);
-    }
-
-    #[test]
-    fn vllm_zmq_appends_tokenizer_eos_ids() {
-        let mut req = vllm_request(vec![], vec![7]);
-        resolve_string_stops(&mut req, Some(&mock_tokenizer()), true);
-        assert_eq!(vllm_params(&req).stop_token_ids, vec![7, 999]);
-
-        // Already-present EOS ids are not duplicated.
-        let mut req = vllm_request(vec![], vec![999]);
-        resolve_string_stops(&mut req, Some(&mock_tokenizer()), true);
-        assert_eq!(vllm_params(&req).stop_token_ids, vec![999]);
-    }
-
-    #[test]
-    fn vllm_zmq_ignore_eos_skips_injection() {
-        let mut req = ProtoGenerateRequest::Vllm(Box::new(vllm_proto::GenerateRequest {
-            sampling_params: Some(vllm_proto::SamplingParams {
-                stop_token_ids: vec![7],
-                ignore_eos: true,
-                ..Default::default()
-            }),
-            ..Default::default()
-        }));
-        resolve_string_stops(&mut req, Some(&mock_tokenizer()), true);
-        assert_eq!(vllm_params(&req).stop_token_ids, vec![7]);
+        assert_eq!(params.stop_token_ids, vec![6]);
     }
 
     #[test]
@@ -774,6 +744,21 @@ mod stop_resolution_tests {
         );
     }
 
+    #[test]
+    fn resolution_returns_router_obligations() {
+        // Strings stripped for the engine come back as the router's trim duty.
+        let mut req = sglang_request(vec![".", "Hello world"], vec![]);
+        let obligations = resolve_string_stops(&mut req, Some(&mock_tokenizer()), false);
+        assert_eq!(
+            obligations,
+            vec![".".to_string(), "Hello world".to_string()]
+        );
+
+        // gRPC vLLM matches stops server-side: nothing left for the router.
+        let mut req = vllm_request(vec!["."], vec![]);
+        assert!(resolve_string_stops(&mut req, Some(&mock_tokenizer()), false).is_empty());
+    }
+
     #[test]
     fn pd_bootstrap_injection_skips_non_sglang_requests() {
         use super::{RuntimeType, Worker, WorkerSelection};

model_gateway/src/routers/grpc/context.rs

@@ -457,6 +457,12 @@ pub(crate) struct ResponseState {
     /// Stop sequence decoder
     pub stop_decoder: Option<StopSequenceDecoder>,
 
+    /// String stops the engine will never match, reported by
+    /// `BackendClient::finalize_generate_request` during request building.
+    /// Response processing must trim these from output text; empty when the
+    /// engine matches stops server-side.
+    pub router_stop_obligations: Vec<String>,
+
     /// Derived skip_special_tokens for streaming (set in preparation, read in response_processing).
     /// Stored here because PreparationOutput is consumed by request_building before
     /// response_processing runs.

model_gateway/src/routers/grpc/harmony/stages/request_building.rs

@@ -396,7 +396,7 @@ impl PipelineStage for HarmonyRequestBuildingStage {
                 };
                 ProtoGenerateRequest::TokenSpeed(Box::new(req))
             }
-            BackendClient::Zmq(_) => {
+            BackendClient::Zmq(zmq_client) if zmq_client.runtime() == RuntimeType::Vllm => {
                 let req = match &ctx.input.request_type {
                     RequestType::Chat(request) => {
                         let body = modified_request
@@ -443,6 +443,17 @@ impl PipelineStage for HarmonyRequestBuildingStage {
                 };
                 ProtoGenerateRequest::Vllm(Box::new(req))
             }
+            // connect() admits only vLLM/TokenSpeed runtimes over ZMQ; a client
+            // reporting anything else is a wiring bug, not a request to serve.
+            BackendClient::Zmq(zmq_client) => {
+                return Err(error::internal_error(
+                    "unsupported_zmq_runtime",
+                    format!(
+                        "ZMQ backend reports unsupported runtime {:?} for Harmony requests",
+                        zmq_client.runtime()
+                    ),
+                ));
+            }
         };
 
         // Inject Harmony stop token IDs into sampling params for ALL Harmony requests
@@ -500,6 +511,12 @@ impl PipelineStage for HarmonyRequestBuildingStage {
             }
         }
 
+        // The client resolves string `stop`s its engine can't match and
+        // reports the router's residual trim obligation; harmony response
+        // processing scans channel text for exactly these strings.
+        ctx.state.response.router_stop_obligations = builder_client
+            .finalize_generate_request(&mut proto_request, ctx.tokenizer_arc().as_ref());
+
         if self.inject_pd_metadata {
             if let Some(workers) = ctx.state.workers.as_ref() {
                 helpers::maybe_inject_pd_metadata(&mut proto_request, workers);

model_gateway/src/routers/grpc/harmony/stages/response_processing.rs

@@ -12,36 +12,18 @@ use crate::{
         error,
         grpc::{
             common::stages::PipelineStage,
-            context::{ClientSelection, FinalResponse, RequestContext, RequestType},
+            context::{FinalResponse, RequestContext, RequestType},
         },
     },
     worker::AttachedBody,
 };
 
-/// String `stop` sequences the ROUTER must enforce: only for direct-ZMQ
-/// backends, where the engine receives token ids and never sees the strings.
-/// Empty for gRPC backends (the engine matches stops itself).
+/// String `stop` sequences the ROUTER must enforce, as reported by the
+/// backend client during request building (its residual obligation: strings
+/// the engine will never match). Empty for engines that match server-side —
+/// no transport inspection here.
 fn router_stop_strings(ctx: &RequestContext) -> Vec<String> {
-    let is_zmq = ctx
-        .state
-        .clients
-        .as_ref()
-        .is_some_and(|clients| match clients {
-            ClientSelection::Single { client } => client.is_zmq(),
-            ClientSelection::Disaggregated { decode, .. } => decode.is_zmq(),
-        });
-    if !is_zmq {
-        return Vec::new();
-    }
-    match &ctx.input.request_type {
-        RequestType::Chat(_) => ctx
-            .chat_request_arc()
-            .stop
-            .as_ref()
-            .map(|stop| stop.to_vec())
-            .unwrap_or_default(),
-        _ => Vec::new(),
-    }
+    ctx.state.response.router_stop_obligations.clone()
 }
 
 /// Harmony Response Processing stage: Parse and format Harmony responses

model_gateway/src/routers/grpc/multimodal/assemble.rs

@@ -112,7 +112,9 @@ async fn assemble_multimodal_data_impl(
             anyhow::bail!("MLX does not support multimodal inputs")
         }
         BackendClient::Zmq(client) => match client.runtime() {
-            RuntimeType::Vllm | RuntimeType::Unspecified => {
+            // connect() admits only vLLM/TokenSpeed runtimes over ZMQ, so no
+            // Unspecified fallback: anything else is a hard error below.
+            RuntimeType::Vllm => {
                 let batch = into_single_batch(intermediate, "vLLM")?;
                 let mut data = assemble_vllm(batch, workers)?;
                 // The ZMQ translate reads tensor bytes inline; this wire has no

model_gateway/src/routers/grpc/regular/stages/chat/request_building.rs

@@ -147,12 +147,11 @@ impl PipelineStage for ChatRequestBuildingStage {
             ctx.state.workers.as_ref(),
         );
 
-        // Resolve string `stop` sequences for engines that can't match them
-        // server-side (SGLang skip_tokenizer_init, and every direct-ZMQ
-        // backend): drop the strings, convert single-token stops to
-        // stop_token_ids; the router-side StopSequenceDecoder trims the text.
-        let is_zmq = builder_client.is_zmq();
-        helpers::resolve_string_stops(&mut proto_request, ctx.tokenizer_arc().as_ref(), is_zmq);
+        // The client resolves string `stop`s its engine can't match and
+        // reports the router's residual trim obligation; no transport
+        // knowledge needed here.
+        ctx.state.response.router_stop_obligations = builder_client
+            .finalize_generate_request(&mut proto_request, ctx.tokenizer_arc().as_ref());
 
         if self.inject_pd_metadata {
             if let Some(workers) = ctx.state.workers.as_ref() {