feat(zmq): typed structured outputs for the direct backend, with e2e coverage

.github/workflows/e2e-gpu-job.yml

@@ -54,6 +54,11 @@ on:
         type: string
         default: "nixl"
         description: "KV transfer backend for vLLM PD workers: nixl or mooncake"
+      connection_mode:
+        required: false
+        type: string
+        default: ""
+        description: "Wire override for local backends: zmq runs the local cases over ZMQ"
 
 jobs:
   run:
@@ -66,6 +71,7 @@ jobs:
       E2E_RUNTIME: ${{ inputs.engine }}
       E2E_GPU_TIER: ${{ inputs.gpu_tier }}
       E2E_VLLM_KV_BACKEND: ${{ inputs.vllm_kv_backend }}
+      E2E_CONNECTION_MODE: ${{ inputs.connection_mode }}
       ROUTER_LOCAL_MODEL_PATH: /models
     steps:
       - name: Checkout code
@@ -166,6 +172,12 @@ jobs:
           if [ "${{ inputs.engine }}" = "vllm" ]; then
             SUFFIX="-${{ inputs.vllm_kv_backend }}"
           fi
+          # gRPC and ZMQ chat legs share engine/GPU/test-dir and differ only by
+          # wire mode; fold it in so their log artifacts don't collide.
+          MODE="${{ inputs.connection_mode }}"
+          if [ -n "$MODE" ]; then
+            SUFFIX="${SUFFIX}-${MODE}"
+          fi
           ARTIFACT="e2e-worker-logs-${{ inputs.engine }}-gpu${{ inputs.gpu_tier }}-${LABEL}${SUFFIX}"
           bash scripts/ci_dump_worker_logs.sh e2e-logs "$ARTIFACT"
           echo "artifact=${ARTIFACT}" >> "$GITHUB_OUTPUT"

.github/workflows/pr-test-rust.yml

@@ -543,6 +543,45 @@ jobs:
       test_dirs: ${{ matrix.test_dirs || 'e2e_test/chat_completions' }}
     secrets: inherit
 
+  e2e-1gpu-chat-zmq:
+    name: e2e-1gpu-chat-zmq (${{ matrix.engine }})
+    needs: [build-wheel, detect-changes]
+    if: >-
+      always()
+      && !cancelled()
+      && needs.build-wheel.result == 'success'
+      && (github.event_name != 'pull_request'
+          || (needs.detect-changes.result == 'success'
+              && (needs.detect-changes.outputs.common == 'true'
+                  || needs.detect-changes.outputs.chat-completions == 'true')))
+    # Same single-worker chat suite as e2e-1gpu-chat, driven over the ZMQ
+    # direct-backend wire. The collection hook deselects the gRPC-only
+    # families (PD, EPD, multi-worker) for a ZMQ lane, so this covers the
+    # local cases only.
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          # The ZMQ lane restarts the engine per model group (66-90s of
+          # load+warmup each); the suite needs more headroom than the gRPC
+          # lane, whose pool amortizes worker bring-up.
+          - engine: vllm
+            timeout: 46
+            test_timeout: 40
+          - engine: tokenspeed
+            timeout: 50
+            test_timeout: 40
+    uses: ./.github/workflows/e2e-gpu-job.yml
+    with:
+      engine: ${{ matrix.engine }}
+      gpu_tier: "1"
+      runner: 1-gpu-h100
+      timeout: ${{ matrix.timeout }}
+      test_timeout: ${{ matrix.test_timeout }}
+      test_dirs: e2e_test/chat_completions
+      connection_mode: zmq
+    secrets: inherit
+
   e2e-1gpu-completions:
     name: e2e-1gpu-completions (${{ matrix.engine }})
     needs: [build-wheel, detect-changes]
@@ -1049,7 +1088,7 @@ jobs:
           path: benchmark_go_bindings/
 
   finish:
-    needs: [pre-commit, python-lint, grpc-proto-build-check, build-wheel, python-unit-tests, unit-tests, benchmarks, e2e-1gpu-chat, e2e-1gpu-completions, e2e-1gpu-embeddings, e2e-1gpu-gateway, e2e-1gpu-responses, e2e-2gpu-pd, e2e-4gpu-chat, e2e-4gpu-gateway, e2e-4gpu-epd, e2e-vendor, go-unit-tests, go-bindings-e2e]
+    needs: [pre-commit, python-lint, grpc-proto-build-check, build-wheel, python-unit-tests, unit-tests, benchmarks, e2e-1gpu-chat, e2e-1gpu-chat-zmq, e2e-1gpu-completions, e2e-1gpu-embeddings, e2e-1gpu-gateway, e2e-1gpu-responses, e2e-2gpu-pd, e2e-4gpu-chat, e2e-4gpu-gateway, e2e-4gpu-epd, e2e-vendor, go-unit-tests, go-bindings-e2e]
     if: always()
     runs-on: k8s-runner-cpu
     permissions: {}
@@ -1064,6 +1103,7 @@ jobs:
                 "${{ needs.unit-tests.result }}" == "failure" || \
                 "${{ needs.benchmarks.result }}" == "failure" || \
                 "${{ needs.e2e-1gpu-chat.result }}" == "failure" || \
+                "${{ needs.e2e-1gpu-chat-zmq.result }}" == "failure" || \
                 "${{ needs.e2e-1gpu-completions.result }}" == "failure" || \
                 "${{ needs.e2e-1gpu-embeddings.result }}" == "failure" || \
                 "${{ needs.e2e-1gpu-gateway.result }}" == "failure" || \

crates/engine_zmq_client/src/codec/tensor.rs

@@ -136,6 +136,42 @@ impl WireNdArray {
         Self::from_raw_bytes(dtype, shape, Bytes::from(data))
     }
 
+    /// Build from little-endian `float32` bytes, casting each element to
+    /// `dtype`. Mirrors the model-dtype cast the engine's own frontend applies
+    /// to floating multimodal tensors before they reach the model.
+    pub fn from_f32_bytes_cast(
+        dtype: super::dtype::ModelDtype,
+        shape: Vec<usize>,
+        data: &[u8],
+    ) -> std::result::Result<Self, String> {
+        use super::dtype::ModelDtype;
+        if !data.len().is_multiple_of(4) {
+            return Err(format!(
+                "float32 buffer length {} is not a multiple of 4",
+                data.len()
+            ));
+        }
+        validate_element_count(&shape, data.len() / 4)?;
+        let floats = data
+            .as_chunks::<4>()
+            .0
+            .iter()
+            .map(|c| f32::from_le_bytes(*c));
+        Ok(match dtype {
+            ModelDtype::Float32 => Self::from_raw("float32", shape, data.to_vec()),
+            ModelDtype::Float16 => Self::from_raw_bytes(
+                "float16",
+                shape,
+                bytes_from_pod_vec(floats.map(f16::from_f32).collect::<Vec<_>>()),
+            ),
+            ModelDtype::BFloat16 => Self::from_raw_bytes(
+                "bfloat16",
+                shape,
+                bytes_from_pod_vec(floats.map(bf16::from_f32).collect::<Vec<_>>()),
+            ),
+        })
+    }
+
     /// Build from an owned immutable raw-view buffer.
     pub fn from_raw_bytes(dtype: impl Into<String>, shape: Vec<usize>, data: Bytes) -> Self {
         Self {

crates/engine_zmq_client/src/error.rs

@@ -53,6 +53,8 @@ pub enum Error {
         context: &'static str,
         field: &'static str,
     },
+    #[error("invalid structured outputs params: {message}")]
+    InvalidStructuredOutputsParams { message: String },
     #[error("request `{request_id}` is already in flight")]
     DuplicateRequestId { request_id: String },
     #[error("data parallel rank {rank} is out of range for {num_engines} engine(s)")]

crates/engine_zmq_client/src/protocol/tokenspeed/sampling.rs

@@ -54,13 +54,14 @@ pub struct SamplingParams {
     pub repetition_penalty: f64,
     /// Minimum number of tokens to generate before EOS / stop handling.
     pub min_new_tokens: u32,
-    /// Structured-output JSON schema. SMG rejects constraints upstream.
+    /// Structured-output JSON schema. Set from the request's `constraint`;
+    /// at most one of these four structured-output fields is populated.
     pub json_schema: Option<String>,
-    /// Structured-output regex. SMG rejects constraints upstream.
+    /// Structured-output regex. Set from the request's `constraint`.
     pub regex: Option<String>,
-    /// Structured-output EBNF grammar. SMG rejects constraints upstream.
+    /// Structured-output EBNF grammar. Set from the request's `constraint`.
     pub ebnf: Option<String>,
-    /// Structured-output structural tag. SMG rejects constraints upstream.
+    /// Structured-output structural tag. Set from the request's `constraint`.
     pub structural_tag: Option<String>,
     /// Ignore the EOS token and keep generating until another stop condition.
     pub ignore_eos: bool,

crates/engine_zmq_client/src/protocol/vllm/mod.rs

@@ -7,20 +7,21 @@
 //! shapes, field order, and `array_like` positional-tuple encoding are the wire
 //! contract with Python `EngineCoreProc` — do not reorder fields.
 //!
-//! Text generation is typed fully. Multimodal features, structured outputs
-//! (guided decoding), and pooling params are carried as [`crate::codec::OpaqueValue`]
-//! for now — they serialize as `nil` on the text path and get strongly typed in
-//! the multimodal phase.
+//! Text generation, structured outputs (guided decoding), and multimodal
+//! features are typed fully. Pooling params and prompt embeds are carried as
+//! [`crate::codec::OpaqueValue`] — they serialize as `nil` on supported paths.
 
 // The startup handshake is engine-neutral (TokenSpeed speaks the same
 // protocol); re-exported here so existing `vllm::handshake` paths keep working.
 pub use crate::protocol::handshake;
 pub mod logprobs;
 pub mod lora;
+pub mod multimodal;
 pub mod output;
 pub mod request;
 pub mod sampling;
 pub mod stats;
+pub mod structured_outputs;
 
 use bytes::Bytes;