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,42 @@ 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:
+          - engine: vllm
+            timeout: 24
+            test_timeout: 18
+          - engine: tokenspeed
+            timeout: 50
+            test_timeout: 18
+    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 +1085,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 +1100,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/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/vllm/mod.rs

@@ -7,10 +7,10 @@
 //! 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 and structured outputs (guided decoding) are typed fully.
+//! Remaining multimodal features 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.
 
 // The startup handshake is engine-neutral (TokenSpeed speaks the same
 // protocol); re-exported here so existing `vllm::handshake` paths keep working.
@@ -21,6 +21,7 @@ pub mod output;
 pub mod request;
 pub mod sampling;
 pub mod stats;
+pub mod structured_outputs;
 
 use bytes::Bytes;
 

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

@@ -2,16 +2,13 @@
 //
 // Ported from the Apache-2.0 reference `vllm-engine-core-client`
 // (vllm-project/vllm): protocol/sampling.rs.
-//
-// `structured_outputs` (guided decoding) is carried as OpaqueValue for now; it
-// serializes as `nil` when absent and gets strongly typed in a later phase.
 
 use std::collections::{BTreeSet, HashMap};
 
 use serde::{Deserialize, Serialize};
 use serde_default::DefaultFromSerde;
 
-use crate::codec::OpaqueValue;
+use super::structured_outputs::StructuredOutputsParams;
 
 fn default_top_p() -> f32 {
     1.0
@@ -115,9 +112,9 @@ pub struct EngineCoreSamplingParams {
     /// Tokenized bad words to avoid during generation.
     #[serde(rename = "_bad_words_token_ids")]
     pub bad_words_token_ids: Option<Vec<Vec<u32>>>,
-    /// Structured outputs (guided decoding). Carried untyped for now; `nil` on
-    /// the text path.
-    pub structured_outputs: Option<OpaqueValue>,
+    /// Structured outputs (guided decoding). `None` (serialized `nil`) on the
+    /// unconstrained text path.
+    pub structured_outputs: Option<StructuredOutputsParams>,
     /// Specific token IDs for which log probabilities should be returned at each
     /// position, in addition to the sampled/scored token.
     pub logprob_token_ids: Option<Vec<u32>>,