[Bugfix][Rust Frontend] Accept appended EngineCoreOutput fields (#56533)
Co-authored-by: Codex <noreply@openai.com> Signed-off-by: Alec Flowers <aflowers@nvidia.com>
Alec committed
Sep 16, 2026 at 18:54 UTC
28ec03bc6e51b17cd9a6b48edc3819e542c2ebb1
5 files changed
+136
rust/src/engine-core-client/src/protocol/mod.rs
+1
@@ -30,6 +30,7 @@ pub mod multimodal;
30
pub mod output;
31
pub mod request;
32
pub mod sampling;
33
+mod serde_utils;
34
pub mod stats;
35
pub mod structured_outputs;
36
pub mod tensor;
rust/src/engine-core-client/src/protocol/output.rs
+60
@@ -9,7 +9,9 @@ use serde::{Deserialize, Deserializer, Serialize, Serializer};
9
use serde_default::DefaultFromSerde;
10
use serde_repr::{Deserialize_repr, Serialize_repr};
11
use serde_tuple::{Deserialize_tuple, Serialize_tuple};
12
+use serde_with::serde_as;
13
14
+use super::serde_utils::AllowTrailingFields;
15
use super::utility::UtilityOutput;
16
use crate::error::{Error, Result, ext_value_decode};
17
use crate::protocol::logprobs::MaybeWireLogprobs;
@@ -156,11 +158,14 @@ impl EngineCoreOutput {
158
///
159
/// Original Python definition:
160
/// <https://github.com/vllm-project/vllm/blob/f22d6e026798a74e6542a52ef776c054f2de572a/vllm/v1/engine/__init__.py#L186-L214>
161
+#[serde_as]
162
#[derive(Debug, Clone, PartialEq, Serialize_tuple, Deserialize_tuple, DefaultFromSerde)]
163
struct WireEngineCoreOutputs {
164
#[serde(default)]
165
engine_index: u32,
166
/// Outputs grouped for this client in the current engine tick.
167
+ // Match msgspec's append-only schema evolution, including Omni extensions.
168
+ #[serde_as(deserialize_as = "Vec<AllowTrailingFields>")]
169
#[serde(default)]
170
outputs: Vec<EngineCoreOutput>,
171
#[serde(default)]
@@ -404,6 +409,61 @@ mod tests {
409
);
410
}
411
412
+ #[test]
413
+ fn engine_core_outputs_ignore_appended_fields() {
414
+ let output = EngineCoreOutput {
415
+ request_id: "req-1".into(),
416
+ new_token_ids: vec![42],
417
+ finish_reason: Some(EngineCoreFinishReason::Length),
418
+ ..Default::default()
419
+ };
420
+ let mut fields = crate::protocol::decode_value(&encode_msgpack(&output).unwrap())
421
+ .unwrap()
422
+ .as_array()
423
+ .unwrap()
424
+ .clone();
425
+ // Omni appends multimodal_output, is_segment_finished, and
426
+ // new_prompt_len_snapshot. Unknown payloads may contain nested values.
427
+ fields.extend([
428
+ OpaqueValue::Map(vec![(
429
+ "audio".into(),
430
+ OpaqueValue::Array(vec![OpaqueValue::Ext(1, vec![0, 1])]),
431
+ )]),
432
+ false.into(),
433
+ OpaqueValue::Nil,
434
+ ]);
435
+ let wire = (0, vec![OpaqueValue::Array(fields); 2]);
436
+ let frames = [Bytes::from(encode_msgpack(&wire).unwrap())];
437
+ let decoded = decode_engine_core_outputs(&frames).unwrap();
438
+ // A second output also checks that the first output's tail was consumed.
439
+ assert_eq!(decoded.as_request_batch().unwrap().outputs, vec![output; 2]);
440
+ }
441
+
442
+ #[test]
443
+ fn engine_core_output_requires_valid_known_fields() {
444
+ for fields in [
445
+ vec![],
446
+ vec!["req-1".into()],
447
+ vec!["req-1".into(), OpaqueValue::Nil],
448
+ vec!["req-1".into(), OpaqueValue::Array(vec![]), false.into()],
449
+ ] {
450
+ let wire = (0, vec![OpaqueValue::Array(fields)]);
451
+ let frames = [Bytes::from(encode_msgpack(&wire).unwrap())];
452
+ assert!(decode_engine_core_outputs(&frames).is_err());
453
+ }
454
+ let wire = (0, vec![("req-1", vec![42_u32])]);
455
+ let frames = [Bytes::from(encode_msgpack(&wire).unwrap())];
456
+ let decoded = decode_engine_core_outputs(&frames).unwrap();
457
+ assert_eq!(
458
+ decoded.as_request_batch().unwrap().outputs,
459
+ vec![EngineCoreOutput {
460
+ request_id: "req-1".into(),
461
+ new_token_ids: vec![42],
462
+ ..Default::default()
463
+ }]
464
+ );
465
+ }
466
+
467
#[test]
468
fn engine_core_outputs_classify_request_batch() {
469
let outputs = WireEngineCoreOutputs {
rust/src/engine-core-client/src/protocol/serde_utils.rs
new
+42
@@ -0,0 +1,42 @@
1
+// SPDX-License-Identifier: Apache-2.0
2
+// SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
+
4
+use std::fmt;
5
+use std::marker::PhantomData;
6
+
7
+use serde::de::{IgnoredAny, SeqAccess, Visitor, value::SeqAccessDeserializer};
8
+use serde::{Deserialize, Deserializer};
9
+use serde_with::DeserializeAs;
10
+
11
+/// Decode an array-like struct using its derived field types and defaults,
12
+/// skipping appended fields as Python's msgspec does. Opt in only for schemas
13
+/// whose extensions preserve the known field order and semantics.
14
+pub(super) struct AllowTrailingFields;
15
+
16
+impl<'de, T: Deserialize<'de>> DeserializeAs<'de, T> for AllowTrailingFields {
17
+ fn deserialize_as<D>(deserializer: D) -> Result<T, D::Error>
18
+ where
19
+ D: Deserializer<'de>,
20
+ {
21
+ struct ArrayVisitor<T>(PhantomData<T>);
22
+
23
+ impl<'de, T: Deserialize<'de>> Visitor<'de> for ArrayVisitor<T> {
24
+ type Value = T;
25
+
26
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
27
+ formatter.write_str("an array-like struct")
28
+ }
29
+
30
+ fn visit_seq<A>(self, mut seq: A) -> Result<T, A::Error>
31
+ where
32
+ A: SeqAccess<'de>,
33
+ {
34
+ let value = T::deserialize(SeqAccessDeserializer::new(&mut seq))?;
35
+ while seq.next_element::<IgnoredAny>()?.is_some() {}
36
+ Ok(value)
37
+ }
38
+ }
39
+
40
+ deserializer.deserialize_seq(ArrayVisitor(PhantomData))
41
+ }
42
+}
rust/src/engine-core-client/src/tests/client.rs
+9
@@ -2641,6 +2641,7 @@ fn python_msgpack_fixtures_match_rust_encoding() {
2641
let multi_connector_stats_hex =
2642
lines.next().expect("missing MultiConnector stats fixture line");
2643
let ready_response_hex = lines.next().expect("missing ready response fixture line");
2644
+ let extended_outputs_hex = lines.next().expect("missing extended outputs fixture line");
2645
2646
let request_bytes = hex::decode(request_hex).unwrap();
2647
let multimodal_request_bytes = hex::decode(multimodal_request_hex).unwrap();
@@ -2722,6 +2723,14 @@ fn python_msgpack_fixtures_match_rust_encoding() {
2723
));
2724
2725
let decoded_outputs: EngineCoreOutputs = rmp_serde::from_slice(&outputs_bytes).unwrap();
2726
+ // Match msgspec's base-schema result for the same extended Python message.
2727
+ let extended_frames = [bytes::Bytes::from(
2728
+ hex::decode(extended_outputs_hex).unwrap(),
2729
+ )];
2730
+ assert_eq!(
2731
+ decode_engine_core_outputs(&extended_frames).unwrap(),
2732
+ decoded_outputs
2733
+ );
2734
expect_test::expect![[r#"
2735
RequestBatch(
2736
RequestBatchOutputs {
rust/src/engine-core-client/src/tests/python_compat.py
+24
@@ -101,6 +101,14 @@ class EngineCoreOutput(
101
num_nans_in_logits: int = 0
102
mm_cache_miss_hashes: list[str] | None = None
103
new_sampling_mask: object | None = None
104
+ spec_decode_metrics: object | None = None
105
+
106
+
107
+class ExtendedEngineCoreOutput(EngineCoreOutput):
108
+ # Match Omni's append-only schema without requiring Omni or torch.
109
+ multimodal_output: dict[str, object] | None = None
110
+ is_segment_finished: bool = False
111
+ new_prompt_len_snapshot: int | None = None
112
113
114
class EngineCoreOutputs(
@@ -207,6 +215,21 @@ outputs = EngineCoreOutputs(
215
finished_requests={"req-1"},
216
)
217
218
+extended_outputs = EngineCoreOutputs(
219
+ outputs=[
220
+ ExtendedEngineCoreOutput(
221
+ **msgspec.structs.asdict(outputs.outputs[0]),
222
+ multimodal_output={"audio": b"\x00\x01"},
223
+ is_segment_finished=True,
224
+ new_prompt_len_snapshot=12,
225
+ )
226
+ ],
227
+ finished_requests=outputs.finished_requests,
228
+)
229
+extended_outputs_bytes = msgspec.msgpack.encode(extended_outputs)
230
+# The ordinary frontend's schema ignores even non-default extension values.
231
+assert msgspec.msgpack.decode(extended_outputs_bytes, type=EngineCoreOutputs) == outputs
232
+
233
sampling_mask_wire = [
234
[
235
"<i4",
@@ -509,3 +532,4 @@ print(msgspec.msgpack.encode(nixl_stats).hex())
532
print(msgspec.msgpack.encode(mooncake_stats).hex())
533
print(msgspec.msgpack.encode(multi_connector_stats).hex())
534
print(msgspec.msgpack.encode(ready_response).hex())
535
+print(extended_outputs_bytes.hex())