feat(zmq): core per-engine dispatch for direct ZMQ backends (1/7) - #2054
Conversation
github-actions
Bot
added
grpc
📝 WalkthroughSummary by CodeRabbit
WalkthroughThe PR adds runtime-specific TokenSpeed and vLLM request handling for ZMQ backends. It updates request builders, metadata responses, sampling translation, logprob streaming, fan-out, validation, and integration tests. ChangesTokenSpeed ZMQ integration
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested labels: Suggested reviewers: Sequence Diagram(s)sequenceDiagram
participant GrpcClient
participant BackendClient
participant ZmqEngineClient
participant RuntimeBackend
GrpcClient->>BackendClient: Build runtime-specific request
BackendClient->>ZmqEngineClient: Forward ProtoGenerateRequest
ZmqEngineClient->>RuntimeBackend: Send translated wire request
RuntimeBackend-->>ZmqEngineClient: Return streamed generation data
ZmqEngineClient-->>BackendClient: Return generation result
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
👋 The PR description doesn't fully follow
Please update the PR description so reviewers have the context they need. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
model_gateway/src/routers/grpc/zmq_client.rs (2)
475-482: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low value🟡 Nit: Avoid the three per-tick
Vecclones in the streaming logprob path.
map_outputruns once per streaming tick, so this is per-token response processing. The chunk clones all three tick vectors, then the originals are moved intostate. You can extendstateby iterator copy and move the tick vectors into the chunk. That removes threeVecallocations per tick.♻️ Proposed refactor
- let chunk_logprobs = (!tick_logprobs_val.is_empty()).then(|| vllm::OutputLogProbs { - token_logprobs: tick_logprobs_val.clone(), - token_ids: tick_logprobs_idx.clone(), - top_logprobs: tick_top_logprobs.clone(), - }); - state.output_logprobs_val.extend(tick_logprobs_val); - state.output_logprobs_idx.extend(tick_logprobs_idx); - state.output_top_logprobs.extend(tick_top_logprobs); + state + .output_logprobs_val + .extend(tick_logprobs_val.iter().copied()); + state + .output_logprobs_idx + .extend(tick_logprobs_idx.iter().copied()); + state + .output_top_logprobs + .extend(tick_top_logprobs.iter().cloned()); + let chunk_logprobs = (!tick_logprobs_val.is_empty()).then(|| vllm::OutputLogProbs { + token_logprobs: tick_logprobs_val, + token_ids: tick_logprobs_idx, + top_logprobs: tick_top_logprobs, + });As per coding guidelines: "Avoid unnecessary clone() calls in gRPC streaming hot paths, especially during per-token response processing."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@model_gateway/src/routers/grpc/zmq_client.rs` around lines 475 - 482, Update the streaming logprob handling around map_output to avoid cloning tick_logprobs_val, tick_logprobs_idx, and tick_top_logprobs: extend the corresponding state vectors by iterating and copying their elements, then move the original tick vectors into the optional vllm::OutputLogProbs chunk. Preserve the existing empty-value behavior and field mappings.Source: Coding guidelines
1158-1294: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win🟡 Nit: Add a test for the engine-error finish reason.
map_outputnow convertsEngineCoreFinishReason::Errorintotonic::Status::internal(lines 484-492). No test covers that path. The mock engine can already send a finish reason, so the case is cheap to add next to this test: sendSome(EngineCoreFinishReason::Error)and assert the stream yields an error item instead of aComplete.As per coding guidelines: "Run the pr-test-analyzer agent to verify that tests adequately cover new or changed functionality."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@model_gateway/src/routers/grpc/zmq_client.rs` around lines 1158 - 1294, The existing generate test does not cover the EngineCoreFinishReason::Error mapping. Add a nearby test using the mock engine setup to send an output with finish_reason set to Error, then assert the generated stream yields an error item containing the expected internal tonic status rather than a Complete response.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@model_gateway/src/routers/grpc/backend_client.rs`:
- Around line 205-231: Update the Self::Zmq(client) branch of is_vllm() to
return whether client.runtime() equals RuntimeType::Vllm, instead of
unconditionally returning true; preserve the existing behavior for other backend
variants.
In `@model_gateway/src/routers/grpc/harmony/stages/request_building.rs`:
- Around line 352-398: Before building the TokenSpeed request in
HarmonyRequestBuildingStage::execute, reject ClientSelection::Disaggregated when
the ZMQ client runtime is RuntimeType::TokenSpeed, returning the existing
bad-request error path used for unsupported Harmony operations. Keep
non-disaggregated TokenSpeed requests and other runtimes unchanged, and ensure
the check occurs before PD execution or request construction.
In `@model_gateway/src/routers/grpc/zmq_client.rs`:
- Around line 880-884: Validate the prompt length before computing
default_max_tokens in the surrounding request translation flow: reject the
request when prompt_len >= max_model_len instead of allowing max_tokens to
become zero. Preserve the existing default calculation for prompts that leave
output capacity, and propagate the rejection as the established context-length
error response.
---
Nitpick comments:
In `@model_gateway/src/routers/grpc/zmq_client.rs`:
- Around line 475-482: Update the streaming logprob handling around map_output
to avoid cloning tick_logprobs_val, tick_logprobs_idx, and tick_top_logprobs:
extend the corresponding state vectors by iterating and copying their elements,
then move the original tick vectors into the optional vllm::OutputLogProbs
chunk. Preserve the existing empty-value behavior and field mappings.
- Around line 1158-1294: The existing generate test does not cover the
EngineCoreFinishReason::Error mapping. Add a nearby test using the mock engine
setup to send an output with finish_reason set to Error, then assert the
generated stream yields an error item containing the expected internal tonic
status rather than a Complete response.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 1185e2a6-d0ae-4f59-9782-343a5133087e
📒 Files selected for processing (5)
crates/grpc_client/src/tokenspeed_scheduler.rsmodel_gateway/src/routers/grpc/backend_client.rsmodel_gateway/src/routers/grpc/client.rsmodel_gateway/src/routers/grpc/harmony/stages/request_building.rsmodel_gateway/src/routers/grpc/zmq_client.rs
💤 Files with no reviewable changes (1)
- crates/grpc_client/src/tokenspeed_scheduler.rs
| BackendClient::Zmq(zmq_client) if zmq_client.runtime() == RuntimeType::TokenSpeed => { | ||
| let req = match &ctx.input.request_type { | ||
| RequestType::Chat(request) => { | ||
| let body = modified_request | ||
| .as_deref() | ||
| .unwrap_or_else(|| request.as_ref()); | ||
| TokenSpeedSchedulerClient::build_generate_request_from_chat( | ||
| request_id, | ||
| body, | ||
| placeholder_processed_text, | ||
| token_ids, | ||
| None, // Harmony path: multimodal not yet wired | ||
| tool_constraints, | ||
| ) | ||
| .map_err(|e| { | ||
| error!(function = "HarmonyRequestBuildingStage::execute", error = %e, "Failed to build TokenSpeed ZMQ generate request"); | ||
| error::bad_request("invalid_request_parameters", format!("Invalid request parameters: {e}")) | ||
| })? | ||
| } | ||
| RequestType::Responses(request) => { | ||
| TokenSpeedSchedulerClient::build_generate_request_from_responses( | ||
| request_id, | ||
| request.as_ref(), | ||
| placeholder_processed_text, | ||
| token_ids, | ||
| tool_constraints, | ||
| ) | ||
| .map_err(|e| { | ||
| error!(function = "HarmonyRequestBuildingStage::execute", error = %e, "Failed to build TokenSpeed ZMQ generate request from responses"); | ||
| error::bad_request("invalid_request_parameters", format!("Invalid request parameters: {e}")) | ||
| })? | ||
| } | ||
| RequestType::Embedding(_) => { | ||
| return Err(error::bad_request( | ||
| "harmony_embedding_not_supported", | ||
| "Embedding requests are not supported with Harmony models".to_string(), | ||
| )); | ||
| } | ||
| _ => { | ||
| return Err(error::bad_request( | ||
| "unsupported_request_type", | ||
| "Unsupported request type for Harmony models".to_string(), | ||
| )); | ||
| } | ||
| }; | ||
| ProtoGenerateRequest::TokenSpeed(Box::new(req)) | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C 5 'PrefillDecode|EncodePrefillDecode|sanitize_sampling_for_prefill|RuntimeType::TokenSpeed' \
model_gateway/src \
-g '*.rs'Repository: smg-project/smg
Length of output: 50372
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate and inspect the request building branch plus nearby client selection path.
wc -l model_gateway/src/routers/grpc/harmony/stages/request_building.rs
sed -n '280,420p' model_gateway/src/routers/grpc/harmony/stages/request_building.rs
# Search for PD mode construction and any TokenSpeed/prefill_decode validation.
rg -n -C 4 'Tokenspeed|TokenSpeed|Tokenspeed|prefill_decode|PD|PrefillDecode|runtime:\|startup_worker_runtime_type|prefill_decode_mode|pd_disaggregation|epd_disaggregation' \
model_gateway/src/config model_gateway/src/workflow model_gateway/src/routers/grpc/harmony \
-g '*.rs'Repository: smg-project/smg
Length of output: 50371
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check whether Harmony PD execution is constructible for TokenSpeed ZMQ clients.
rg -n -C 4 'BuildClientSelection|ClientSelection|prefill_decode|EncodePrefillDecode|backend_type|RuntimeType::TokenSpeed|sanitize_sampling' \
model_gateway/src/routers/grpc/harmony \
-g '*.rs'
# Check whether PD mode is allowed outside Harmony for TokenSpeed ZMQ, and whether routing builds execute in PD.
rg -n -C 6 'PrefillDecode|EncodePrefillDecode|RoutingMode::|is_pd_mode|startup_worker_runtime_type|Backend::Tokenspeed|Tokenspeed' \
model_gateway/src/routers/grpc ../model_gateway/src 2>/dev/null || trueRepository: smg-project/smg
Length of output: 50371
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect workflow/build-time PD mode construction around startup_worker_runtime_type.
wc -l model_gateway/src/main.rs
sed -n '1232,1360p' model_gateway/src/main.rs
sed -n '1360,1445p' model_gateway/src/main.rs
# Inspect worker selection/registration behavior for backends that can support PD URLs plus startup runtime.
rg -n -C 5 'startup_worker_runtime_type|Backend::Tokenspeed|RuntimeType::TokenSpeed|backend_type|connection_mode|WorkerRegistry::.*register|register\(' \
model_gateway/src/workflow model_gateway/src/workflow/steps model_gateway/src/workflow/steps/local model_gateway/src/workflow/steps/k8s \
-g '*.rs'Repository: smg-project/smg
Length of output: 50372
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect WorkerSelectionStage and ClientAcquisitionStage PD handling.
rg -n -C 8 'WorkerSelectionStage|ClientAcquisitionStage|BuildClientSelection|WorkerSelection|ClientSelection|prefill_decode_mode|PrefillDecode|EncodePrefillDecode|worker_type|runtime_type' \
model_gateway/src/routers/grpc/model_gateway/src/routers/grpc/harmony/stages -g '*.rs' 2>/dev/null || true
# Inspect exact mode/pipeline registration for PD/Harmony and CLI PD flags combined with ZMQ.
rg -n -C 4 'CreateStartupWorkersStep|CreateWorkerRegistrationWorkflow|worker_startup_mode|work_dir|PrefillDecode|EncodePrefillDecode|epd_disaggregation|pd_disaggregation|backend_type|Backend::' \
model_gateway/src/workflow model_gateway/src/main.rs \
-g '*.rs'Repository: smg-project/smg
Length of output: 12960
Reject PD selection for TokenSpeed ZMQ requests.
PrefillDecode Harmony pipelines use ClientSelection::Disaggregated, but ProtoGenerateRequest::sanitize_sampling_for_prefill is a no-op for TokenSpeed requests. This lets the prefill leg carry normal sampling instead of a prefill-only handoff payload. Add a runtime-specific check before PD execution; if TokenSpeed ZMQ cannot support PD, treat ClientSelection::Disaggregated as an error for RuntimeType::TokenSpeed.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@model_gateway/src/routers/grpc/harmony/stages/request_building.rs` around
lines 352 - 398, Before building the TokenSpeed request in
HarmonyRequestBuildingStage::execute, reject ClientSelection::Disaggregated when
the ZMQ client runtime is RuntimeType::TokenSpeed, returning the existing
bad-request error path used for unsupported Harmony operations. Keep
non-disaggregated TokenSpeed requests and other runtimes unchanged, and ensure
the check occurs before PD execution or request construction.
Source: Coding guidelines
| // vLLM's frontend defaults an unset `max_tokens` to the remaining context | ||
| // (`max_model_len - prompt_len`). | ||
| let prompt_len = prompt_token_ids.as_ref().map_or(0, |ids| ids.len()) as u64; | ||
| let default_max_tokens = | ||
| u32::try_from(max_model_len.saturating_sub(prompt_len)).unwrap_or(u32::MAX); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🔴 Important: A prompt that fills the context yields max_tokens = 0 and an empty successful completion.
saturating_sub returns 0 when prompt_len >= max_model_len. translate_sampling then sets max_tokens: 0. The engine accepts the request, generates nothing, and finishes with length. The client receives a 200 with no content instead of a context-length error. vLLM's OpenAI frontend, which this path bypasses, rejects the request in that case.
Reject the request when the prompt leaves no room for output.
🐛 Proposed fix
let prompt_len = prompt_token_ids.as_ref().map_or(0, |ids| ids.len()) as u64;
+ if prompt_len >= max_model_len {
+ return Err(format!(
+ "prompt is {prompt_len} tokens, which leaves no room in the model context \
+ of {max_model_len} tokens"
+ ));
+ }
let default_max_tokens =
u32::try_from(max_model_len.saturating_sub(prompt_len)).unwrap_or(u32::MAX);As per coding guidelines: "Run the silent-failure-hunter agent on changed files to detect swallowed errors, inappropriate fallbacks, and missing error propagation."
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // vLLM's frontend defaults an unset `max_tokens` to the remaining context | |
| // (`max_model_len - prompt_len`). | |
| let prompt_len = prompt_token_ids.as_ref().map_or(0, |ids| ids.len()) as u64; | |
| let default_max_tokens = | |
| u32::try_from(max_model_len.saturating_sub(prompt_len)).unwrap_or(u32::MAX); | |
| // vLLM's frontend defaults an unset `max_tokens` to the remaining context | |
| // (`max_model_len - prompt_len`). | |
| let prompt_len = prompt_token_ids.as_ref().map_or(0, |ids| ids.len()) as u64; | |
| if prompt_len >= max_model_len { | |
| return Err(format!( | |
| "prompt is {prompt_len} tokens, which leaves no room in the model context \ | |
| of {max_model_len} tokens" | |
| )); | |
| } | |
| let default_max_tokens = | |
| u32::try_from(max_model_len.saturating_sub(prompt_len)).unwrap_or(u32::MAX); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@model_gateway/src/routers/grpc/zmq_client.rs` around lines 880 - 884,
Validate the prompt length before computing default_max_tokens in the
surrounding request translation flow: reject the request when prompt_len >=
max_model_len instead of allowing max_tokens to become zero. Preserve the
existing default calculation for prompts that leave output capacity, and
propagate the rejection as the established context-length error response.
Source: Coding guidelines
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
There was a problem hiding this comment.
Clean PR — the per-engine dispatch is well-structured and consistently applied across all request builders (chat, messages, completion, generate) and the Harmony pipeline. The native TokenSpeed proto usage eliminates the vLLM-proto re-mapping, and the new top_logprobs shaping + max_model_len defaulting for the vLLM path are solid additions. Test coverage is thorough. No bugs found.
Summary: 0 🔴 Important · 0 🟡 Nit · 0 🟣 Pre-existing
Add the text-only dispatch layer for direct same-host ZMQ engine connections, as a first-class sibling of the gRPC path. - BackendClient gains per-runtime request building: a ZMQ backend builds native vLLM EngineCore or TokenSpeed requests, mirroring the gRPC per-engine dispatch. - zmq_client translates vLLM/TokenSpeed proto requests onto their native wire formats, defaults unset max_tokens to the remaining context, shapes top_logprobs to the requested count, and fans out n>1 into single-sample sub-requests. - TokenSpeedSchedulerClient build_* methods become associated functions so both the gRPC and ZMQ paths can call them without a live client; callers updated accordingly. Structured outputs, multimodal, and EOS/stop forwarding are layered on in follow-up changes; this path rejects those inputs for now. Signed-off-by: Simo Lin <25425177+slin1237@users.noreply.github.com>
79d8f20 to
c9775bc
Compare
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
There was a problem hiding this comment.
🧹 Nitpick comments (2)
model_gateway/src/routers/grpc/zmq_client.rs (2)
475-482: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win🟡 Nit: Remove the per-tick clones in the streaming logprob path.
chunk_logprobsclones all three tick vectors, then the same vectors move intostate. This runs on every decode tick, so each streamed token allocates and copies three vectors twice. Build the chunk from the moved vectors and extendstatefrom its slices instead.♻️ Proposed refactor
- let chunk_logprobs = (!tick_logprobs_val.is_empty()).then(|| vllm::OutputLogProbs { - token_logprobs: tick_logprobs_val.clone(), - token_ids: tick_logprobs_idx.clone(), - top_logprobs: tick_top_logprobs.clone(), - }); - state.output_logprobs_val.extend(tick_logprobs_val); - state.output_logprobs_idx.extend(tick_logprobs_idx); - state.output_top_logprobs.extend(tick_top_logprobs); + let chunk_logprobs = (!tick_logprobs_val.is_empty()).then(|| vllm::OutputLogProbs { + token_logprobs: tick_logprobs_val, + token_ids: tick_logprobs_idx, + top_logprobs: tick_top_logprobs, + }); + if let Some(chunk) = chunk_logprobs.as_ref() { + state + .output_logprobs_val + .extend_from_slice(&chunk.token_logprobs); + state.output_logprobs_idx.extend_from_slice(&chunk.token_ids); + state + .output_top_logprobs + .extend_from_slice(&chunk.top_logprobs); + }As per coding guidelines: "Avoid unnecessary clone() calls in gRPC streaming hot paths, especially during per-token response processing."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@model_gateway/src/routers/grpc/zmq_client.rs` around lines 475 - 482, Update the streaming logprob handling around chunk_logprobs to move the tick vectors into the chunk instead of cloning them, then extend state.output_logprobs_val, state.output_logprobs_idx, and state.output_top_logprobs from the moved vectors’ slices. Preserve the existing empty-check and output contents while eliminating per-tick clone allocations.Source: Coding guidelines
484-492: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win🟡 Nit: Add a test for the
Errorfinish reason mapping.This arm converts an engine-side request failure into
Status::internal. Before this change the same tick produced a normalCompletewith empty output. No test in the file drivesEngineCoreFinishReason::Error, so a regression back to the empty-completion behavior would pass. The existingbatchhelper already accepts a finish reason, so the test is small.As per coding guidelines: "Run the pr-test-analyzer agent to verify that tests adequately cover new or changed functionality."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@model_gateway/src/routers/grpc/zmq_client.rs` around lines 484 - 492, Add a focused test alongside the existing gRPC ZMQ client tests that uses the batch helper with EngineCoreFinishReason::Error, invokes the response handling path, and asserts it returns tonic::Status::internal rather than a normal completion with empty output. Ensure the test verifies the expected error mapping and prevents regression to empty successful responses.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@model_gateway/src/routers/grpc/zmq_client.rs`:
- Around line 475-482: Update the streaming logprob handling around
chunk_logprobs to move the tick vectors into the chunk instead of cloning them,
then extend state.output_logprobs_val, state.output_logprobs_idx, and
state.output_top_logprobs from the moved vectors’ slices. Preserve the existing
empty-check and output contents while eliminating per-tick clone allocations.
- Around line 484-492: Add a focused test alongside the existing gRPC ZMQ client
tests that uses the batch helper with EngineCoreFinishReason::Error, invokes the
response handling path, and asserts it returns tonic::Status::internal rather
than a normal completion with empty output. Ensure the test verifies the
expected error mapping and prevents regression to empty successful responses.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 6bcd7473-f21a-4d70-b276-eff05b4c77ce
📒 Files selected for processing (5)
crates/grpc_client/src/tokenspeed_scheduler.rsmodel_gateway/src/routers/grpc/backend_client.rsmodel_gateway/src/routers/grpc/client.rsmodel_gateway/src/routers/grpc/harmony/stages/request_building.rsmodel_gateway/src/routers/grpc/zmq_client.rs
💤 Files with no reviewable changes (1)
- crates/grpc_client/src/tokenspeed_scheduler.rs
🚧 Files skipped from review as they are similar to previous changes (3)
- model_gateway/src/routers/grpc/client.rs
- model_gateway/src/routers/grpc/harmony/stages/request_building.rs
- model_gateway/src/routers/grpc/backend_client.rs
1 participant
Description
Problem
The direct-ZMQ engine backend (originally #2041) landed as one oversized PR that
mixed the transport skeleton with structured outputs, multimodal, EOS handling,
and Harmony stop matching. That size made it hard to review and risky to merge.
Solution
Split the feature into a dependency-ordered stack. This first PR carries the
text-only core: per-engine (vLLM EngineCore / TokenSpeed) request dispatch behind
the existing vLLM gRPC client surface. Structured outputs, multimodal, and stop
handling are left as typed placeholders that later PRs in the stack fill in and
wire up.
Changes
crates/engine_zmq_client/src/protocol/vllm/{mod,sampling}.rs— text-generationrequest/sampling translation for the vLLM EngineCore wire protocol.
crates/engine_zmq_client/src/error.rs— client error surface for the adapter.model_gateway/src/routers/grpc/zmq_client.rs— per-engine dispatch adaptertranslating the gRPC request into the engine's ZMQ request (text path).
Test Plan
cargo +nightly fmt -- --checkandcargo clippy --all-featuresclean (CIunit-testslane,Run fmt+Run lintsteps).cargo testunit coverage for request/sampling translation (protocol crate).Checklist
cargo +nightly fmtpassescargo clippy --all-targets --all-features -- -D warningspassesStack (split of the oversized #2041, merge bottom-up):
zmq/01-core-dispatch— core per-engine dispatch (this PR)zmq/02-structured-outputs— structured outputszmq/03-multimodal— multimodal inputszmq/04-eos-forwarding— EOS + string-stop resolutionzmq/05-harmony-stop-matcher— Harmony stop stringszmq/06-e2e-infra— e2e harness supportzmq/07-e2e-tests— e2e tests + CI lanes