From f9a419f988868722b4b0b95f1649889caa617ff6 Mon Sep 17 00:00:00 2001 From: Lloyd Mak Date: Thu, 27 Aug 2026 01:43:47 +0000 Subject: [PATCH 1/3] Tolerate a Chutes stream chunk that omits `id` Reproduced against production: streaming zai-org/GLM-5.1-FP8 with a large input context falls back to Chutes, whose decrypted frames omit `id`. The shared ChatCompletionChunk requires it, so a single missing scalar killed an otherwise healthy stream: data: {"error":{"message":"Failed to perform completion: Chutes stream chunk parse: missing field `id` at line 1 column 184", ...}} data: [DONE] Delivered as HTTP 200 with the error in-band and [DONE] after it, so a client keying on [DONE] recorded a successful empty completion. At 96K the same request returns a clean 400, so this path is a defect rather than a limit. The fix is contained to the Chutes boundary. ChatCompletionChunk is shared by every provider; making `id` optional there would weaken validation fleet-wide and let a genuinely malformed chunk from any backend parse silently. Instead the frame is deserialized to a Value, a missing `id` is filled, and the typed chunk is built from that. Unknown fields already survive via the flatten map - verified by test, since DeepSeek emits `prompt_text` at 400K context. `raw_bytes` are unchanged and asserted byte-identical: those bytes are signature-relevant and must pass through verbatim. Handles the consequence: extract_inference_id_from_chunk previously hashed whatever it was given, so an empty id produced the *same* inference id for every affected stream - worse than none, because it looks valid and would corrupt attestation lookups. It now returns Option and absent stays absent, matching how Inference-Id is already omitted when a stream fails before its first chunk. This inverts an existing unit test that asserted the old behaviour; the assertion was encoding the bug. Verified: cargo clippy -p inference_providers -p services -p api --all-targets -D warnings clean; cargo test -p inference_providers --lib chutes 112 passed; cargo test -p api --lib extract_inference_id 4 passed. E2E tests were not run - they require PostgreSQL and a dstack/TEE socket, neither available here. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Fm9f3maaY2wd6yHH8FEjR5 --- crates/api/src/routes/completions.rs | 25 ++++---- .../src/attested/chutes/e2ee_stream.rs | 59 ++++++++++++++++++- 2 files changed, 72 insertions(+), 12 deletions(-) diff --git a/crates/api/src/routes/completions.rs b/crates/api/src/routes/completions.rs index f1aaa7759..cfb99683b 100644 --- a/crates/api/src/routes/completions.rs +++ b/crates/api/src/routes/completions.rs @@ -562,12 +562,12 @@ fn build_final_usage_chunk_bytes( } // Helper function to extract inference ID from a parsed stream chunk -fn extract_inference_id_from_chunk(chunk: &inference_providers::StreamChunk) -> Uuid { +fn extract_inference_id_from_chunk(chunk: &inference_providers::StreamChunk) -> Option { let id = match chunk { inference_providers::StreamChunk::Chat(c) => &c.id, inference_providers::StreamChunk::Text(c) => &c.id, }; - hash_inference_id_to_uuid(id) + (!id.is_empty()).then(|| hash_inference_id_to_uuid(id)) } // Convert MessageContent to serde_json::Value, preserving multimodal parts (images, audio, etc.) @@ -1547,8 +1547,9 @@ async fn chat_completions_inner( stream_chat_id = Some(match chunk { inference_providers::StreamChunk::Chat(c) => c.id.clone(), inference_providers::StreamChunk::Text(c) => c.id.clone(), - }); - break Some(extract_inference_id_from_chunk(chunk)); + }) + .filter(|id| !id.is_empty()); + break extract_inference_id_from_chunk(chunk); } true } @@ -2406,8 +2407,9 @@ async fn completions_inner( stream_chat_id = Some(match chunk { inference_providers::StreamChunk::Chat(c) => c.id.clone(), inference_providers::StreamChunk::Text(c) => c.id.clone(), - }); - break Some(extract_inference_id_from_chunk(chunk)); + }) + .filter(|id| !id.is_empty()); + break extract_inference_id_from_chunk(chunk); } true } @@ -3673,13 +3675,14 @@ mod tests { #[test] fn test_extract_inference_id_from_chunk_empty_id() { + // Given: Chutes omitted its provider inference id at the parsing boundary. let chunk = make_chat_chunk(""); + + // When: the route derives the optional public inference id. let result = extract_inference_id_from_chunk(&chunk); - // Empty string should still produce a valid UUID - assert!( - !result.is_nil(), - "empty provider ID should still produce a non-nil UUID" - ); + + // Then: absence stays absent instead of becoming a hash of the empty string. + assert_eq!(result, None); } fn empty_delta() -> inference_providers::models::ChatDelta { diff --git a/crates/inference_providers/src/attested/chutes/e2ee_stream.rs b/crates/inference_providers/src/attested/chutes/e2ee_stream.rs index 3bfdba29b..a4f8a4048 100644 --- a/crates/inference_providers/src/attested/chutes/e2ee_stream.rs +++ b/crates/inference_providers/src/attested/chutes/e2ee_stream.rs @@ -84,7 +84,14 @@ fn inner_event(plaintext: &[u8]) -> Result, CompletionError> { if content == "[DONE]" { return Ok(Some(done_event())); } - let chunk: crate::ChatCompletionChunk = serde_json::from_str(content) + let mut chunk_json: serde_json::Value = serde_json::from_str(content) + .map_err(|e| CompletionError::CompletionError(format!("Chutes stream chunk parse: {e}")))?; + if let Some(object) = chunk_json.as_object_mut() { + object + .entry("id") + .or_insert_with(|| serde_json::Value::String(String::new())); + } + let chunk: crate::ChatCompletionChunk = serde_json::from_value(chunk_json) .map_err(|e| CompletionError::CompletionError(format!("Chutes stream chunk parse: {e}")))?; Ok(Some(SSEEvent { // Hand clients a clean, well-framed OpenAI SSE line. @@ -258,6 +265,56 @@ mod tests { assert!(inner_event(line).unwrap().is_some()); } + #[test] + fn inner_event_accepts_missing_id_and_preserves_raw_bytes() { + // Given: a valid Chutes frame whose provider-specific shape omits `id`. + let frame = concat!( + "data: ", + r#"{"object":"chat.completion.chunk","created":0,"model":"m","choices":[]}"#, + "\n\n" + ); + + // When: the decrypted frame crosses the Chutes-only parser boundary. + let event = inner_event(frame.as_bytes()).unwrap().unwrap(); + + // Then: the typed chunk is usable without inventing an id, and signed bytes are unchanged. + let Some(StreamChunk::Chat(chunk)) = event.chunk else { + panic!("expected a parsed chat chunk"); + }; + assert!(chunk.id.is_empty()); + assert_eq!(event.raw_bytes.as_ref(), frame.as_bytes()); + } + + #[test] + fn inner_event_preserves_unknown_top_level_fields() { + // Given: a valid provider frame with a field outside the shared schema. + let frame = br#"{"id":"x","object":"chat.completion.chunk","created":0,"model":"m","choices":[],"prompt_text":null}"#; + + // When: the frame is parsed at the Chutes boundary. + let event = inner_event(frame).unwrap().unwrap(); + + // Then: the shared chunk's flatten map preserves the provider field. + let Some(StreamChunk::Chat(chunk)) = event.chunk else { + panic!("expected a parsed chat chunk"); + }; + assert_eq!( + chunk.extra.get("prompt_text"), + Some(&serde_json::Value::Null) + ); + } + + #[test] + fn inner_event_rejects_missing_choices() { + // Given: JSON that is not a valid completion chunk because `choices` is absent. + let frame = br#"{"id":"x","object":"chat.completion.chunk","created":0,"model":"m"}"#; + + // When: it crosses the Chutes parser boundary. + let error = inner_event(frame).unwrap_err(); + + // Then: tolerance for a missing id does not weaken required completion data. + assert!(format!("{error}").contains("missing field `choices`")); + } + #[test] fn inner_event_done_and_empty() { assert!(inner_event(b"data: [DONE]") From 1926b783ffb937647bdf645b8a5b2e07bebaebe6 Mon Sep 17 00:00:00 2001 From: Lloyd Mak Date: Thu, 27 Aug 2026 20:32:50 +0000 Subject: [PATCH 2/3] Address review: keep raw and typed chunks in sync, mint a per-stream id Two review findings, both about consequences downstream of the parse. High: the id placeholder was inserted only into the value used to build the typed chunk, so raw_bytes stayed id-less. rewrite_sse_event_model reads raw_bytes, sanitizes against the Chutes allowlist and applies the canonical model rewrite, then rebuilds the typed chunk from it - and that rebuild failed on an id-less frame, so it kept the ORIGINAL chunk. The attested path then reserializes that stale chunk for usage shaping, which can re-expose fields the allowlist strips (prompt_token_ids, prompt_sha256) and loses the canonical model rewrite. The tolerance could therefore leak provider internals on exactly the streams it unblocks. Medium: an empty id became the signature key. Gateway signatures upsert on (chat_id, signing_algo), so every affected completion overwrote the previous one's row and no per-completion signature survived. Both are fixed by minting a stable synthetic id per response session - `chutes-gateway-` - used when the frame omits `id` or carries an empty one. It is stable across every chunk of a stream so chat_id grouping, sticky routing and signature storage all behave; unique per completion so signature rows cannot collide; and visibly synthetic so it cannot be mistaken for a provider id. Raw and typed representations are both built from the value carrying that id, so every downstream round-trip through raw_bytes succeeds. Re-serialization is deliberately conditional. serde_json here has no preserve_order feature, so Value::Object is a BTreeMap and a round-trip sorts keys alphabetically. Re-serializing unconditionally would have changed the wire bytes of every Chutes frame, not just the id-less ones this PR exists to fix. Frames that already carry an id are emitted byte-for-byte as before, guarded by a regression test that fails against the unconditional version. Verified: cargo clippy -p inference_providers --all-targets -D warnings clean; cargo test -p inference_providers --lib chutes 116 passed, 0 failed. The sanitization regression test fails without the fix. E2E not run - needs PostgreSQL and a dstack/TEE socket. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Fm9f3maaY2wd6yHH8FEjR5 --- .../src/attested/chutes/e2ee.rs | 10 +- .../src/attested/chutes/e2ee_stream.rs | 225 ++++++++++++++++-- 2 files changed, 212 insertions(+), 23 deletions(-) diff --git a/crates/inference_providers/src/attested/chutes/e2ee.rs b/crates/inference_providers/src/attested/chutes/e2ee.rs index fc089367d..ef3f2f77a 100644 --- a/crates/inference_providers/src/attested/chutes/e2ee.rs +++ b/crates/inference_providers/src/attested/chutes/e2ee.rs @@ -166,6 +166,7 @@ pub struct PreparedRequest { /// so this must outlive the whole response. pub struct ResponseSession { response_dk: ml_kem::DecapsulationKey768, + synthetic_stream_id: String, } /// Build an E2EE request blob for `request_json` (the OpenAI request body) @@ -215,11 +216,18 @@ pub fn build_request( Ok(PreparedRequest { blob, - session: ResponseSession { response_dk }, + session: ResponseSession { + response_dk, + synthetic_stream_id: format!("chutes-gateway-{}", uuid::Uuid::new_v4()), + }, }) } impl ResponseSession { + pub(super) fn synthetic_stream_id(&self) -> &str { + &self.synthetic_stream_id + } + /// Decrypt a non-streaming response blob (`mlkem_ct ‖ nonce ‖ ct ‖ tag`, /// keyed with `info="e2e-resp-v1"`, gzip-compressed) into the OpenAI /// response JSON bytes. diff --git a/crates/inference_providers/src/attested/chutes/e2ee_stream.rs b/crates/inference_providers/src/attested/chutes/e2ee_stream.rs index a4f8a4048..50587f603 100644 --- a/crates/inference_providers/src/attested/chutes/e2ee_stream.rs +++ b/crates/inference_providers/src/attested/chutes/e2ee_stream.rs @@ -66,7 +66,10 @@ fn done_event() -> SSEEvent { /// Parse one *decrypted* plaintext frame (a raw OpenAI SSE line, e.g. /// `data: {chunk}` or bare `{chunk}`) into an [`SSEEvent`]. Returns `None` for an /// empty frame. Pure — unit-tested without any crypto. -fn inner_event(plaintext: &[u8]) -> Result, CompletionError> { +fn inner_event( + plaintext: &[u8], + synthetic_stream_id: &str, +) -> Result, CompletionError> { let s = String::from_utf8_lossy(plaintext); let s = s.trim(); // An SSE comment / keepalive line (e.g. `: ping`) — vLLM/SGLang backends emit @@ -86,16 +89,36 @@ fn inner_event(plaintext: &[u8]) -> Result, CompletionError> { } let mut chunk_json: serde_json::Value = serde_json::from_str(content) .map_err(|e| CompletionError::CompletionError(format!("Chutes stream chunk parse: {e}")))?; - if let Some(object) = chunk_json.as_object_mut() { - object - .entry("id") - .or_insert_with(|| serde_json::Value::String(String::new())); - } + let needs_synthetic_id = if let Some(object) = chunk_json.as_object_mut() { + let needs_synthetic_id = match object.get("id") { + None => true, + Some(serde_json::Value::String(id)) => id.is_empty(), + Some(_) => false, + }; + if needs_synthetic_id { + object.insert( + "id".to_string(), + serde_json::Value::String(synthetic_stream_id.to_string()), + ); + } + needs_synthetic_id + } else { + false + }; + let normalized_content; + let raw_content = if needs_synthetic_id { + normalized_content = serde_json::to_string(&chunk_json).map_err(|e| { + CompletionError::CompletionError(format!("Chutes stream chunk parse: {e}")) + })?; + normalized_content.as_str() + } else { + content + }; let chunk: crate::ChatCompletionChunk = serde_json::from_value(chunk_json) .map_err(|e| CompletionError::CompletionError(format!("Chutes stream chunk parse: {e}")))?; Ok(Some(SSEEvent { // Hand clients a clean, well-framed OpenAI SSE line. - raw_bytes: Bytes::from(format!("data: {content}\n\n")), + raw_bytes: Bytes::from(format!("data: {raw_content}\n\n")), chunk: Some(StreamChunk::Chat(chunk)), raw_passthrough: true, })) @@ -153,7 +176,7 @@ fn handle_outer_payload( let plaintext = key.decrypt_chunk(&frame).map_err(|e| { CompletionError::CompletionError(format!("Chutes stream chunk decrypt: {e}")) })?; - inner_event(&plaintext) + inner_event(&plaintext, session.synthetic_stream_id()) } else if let Some(err) = obj.get("e2e_error").and_then(|x| x.as_str()) { Err(CompletionError::CompletionError(format!( "Chutes stream error: {err}" @@ -235,6 +258,8 @@ mod tests { use ml_kem::kem::{Kem, KeyExport}; use ml_kem::MlKem768; + const SYNTHETIC_STREAM_ID: &str = "chutes-gateway-test"; + fn fresh_session() -> ResponseSession { // A valid instance pubkey so build_request succeeds; we only exercise the // non-crypto control paths ([DONE], e2e_error) with the returned session. @@ -253,20 +278,39 @@ mod tests { #[test] fn inner_event_parses_data_prefixed_chunk() { let line = b"data: {\"id\":\"x\",\"object\":\"chat.completion.chunk\",\"created\":0,\"model\":\"m\",\"choices\":[]}"; - let ev = inner_event(line).unwrap().unwrap(); + let ev = inner_event(line, SYNTHETIC_STREAM_ID).unwrap().unwrap(); assert!(matches!(ev.chunk, Some(StreamChunk::Chat(_)))); assert!(ev.raw_passthrough); assert!(ev.raw_bytes.starts_with(b"data: ")); } + #[test] + fn inner_event_preserves_raw_bytes_when_id_is_present() { + // Given: a valid provider frame whose key order and whitespace differ + // from serde_json's normalized representation. + let frame = concat!( + "data: ", + r#"{"model": "m", "id": "provider-id", "object": "chat.completion.chunk", "created": 0, "choices": []}"#, + "\n\n" + ); + + // When: the frame crosses the Chutes parser without needing an id. + let event = inner_event(frame.as_bytes(), SYNTHETIC_STREAM_ID) + .unwrap() + .unwrap(); + + // Then: passthrough bytes remain exactly as the provider emitted them. + assert_eq!(event.raw_bytes.as_ref(), frame.as_bytes()); + } + #[test] fn inner_event_parses_bare_json_chunk() { let line = b"{\"id\":\"x\",\"object\":\"chat.completion.chunk\",\"created\":0,\"model\":\"m\",\"choices\":[]}"; - assert!(inner_event(line).unwrap().is_some()); + assert!(inner_event(line, SYNTHETIC_STREAM_ID).unwrap().is_some()); } #[test] - fn inner_event_accepts_missing_id_and_preserves_raw_bytes() { + fn inner_event_accepts_missing_id_and_normalizes_raw_bytes() { // Given: a valid Chutes frame whose provider-specific shape omits `id`. let frame = concat!( "data: ", @@ -275,14 +319,147 @@ mod tests { ); // When: the decrypted frame crosses the Chutes-only parser boundary. - let event = inner_event(frame.as_bytes()).unwrap().unwrap(); + let event = inner_event(frame.as_bytes(), SYNTHETIC_STREAM_ID) + .unwrap() + .unwrap(); - // Then: the typed chunk is usable without inventing an id, and signed bytes are unchanged. + // Then: the typed and reconstructed raw representations are semantically + // equivalent and carry the same injected stream id. let Some(StreamChunk::Chat(chunk)) = event.chunk else { panic!("expected a parsed chat chunk"); }; - assert!(chunk.id.is_empty()); - assert_eq!(event.raw_bytes.as_ref(), frame.as_bytes()); + assert_eq!(chunk.id, SYNTHETIC_STREAM_ID); + + let raw_line = std::str::from_utf8(&event.raw_bytes).unwrap(); + let raw: serde_json::Value = serde_json::from_str( + raw_line + .strip_prefix("data: ") + .expect("normalized event is a framed SSE data line") + .trim_end(), + ) + .unwrap(); + let mut expected: serde_json::Value = serde_json::from_str( + frame + .strip_prefix("data: ") + .expect("fixture is a framed SSE data line") + .trim_end(), + ) + .unwrap(); + expected["id"] = serde_json::Value::String(SYNTHETIC_STREAM_ID.to_string()); + assert_eq!(raw, expected); + } + + #[test] + fn inner_event_replaces_empty_id_in_raw_and_typed_chunk() { + // Given: a valid Chutes frame whose provider id is an empty string. + let frame = + br#"{"id":"","object":"chat.completion.chunk","created":0,"model":"m","choices":[]}"#; + + // When: the decrypted frame crosses the Chutes-only parser boundary. + let event = inner_event(frame, SYNTHETIC_STREAM_ID).unwrap().unwrap(); + + // Then: both emitted representations carry the same synthetic id. + let raw: serde_json::Value = serde_json::from_slice( + event + .raw_bytes + .strip_prefix(b"data: ") + .expect("normalized event is a framed SSE data line"), + ) + .unwrap(); + let Some(StreamChunk::Chat(chunk)) = event.chunk else { + panic!("expected a parsed chat chunk"); + }; + assert_eq!(raw["id"], SYNTHETIC_STREAM_ID); + assert_eq!(chunk.id, SYNTHETIC_STREAM_ID); + } + + #[test] + fn missing_id_raw_and_typed_round_trip_sanitizes_provider_fields() { + // Given: an id-less provider frame carrying fields that the Chutes + // client-facing allowlist must remove. + let frame = br#"{"object":"chat.completion.chunk","created":0,"model":"provider/model","prompt_token_ids":[1,2],"prompt_sha256":"secret","choices":[]}"#; + let session = fresh_session(); + let event = inner_event(frame, session.synthetic_stream_id()) + .unwrap() + .unwrap(); + + // When: the downstream Chutes rewrite sanitizes and canonicalizes it. + let rewritten = + super::super::rewrite_sse_event_model(event, Some("canonical/model"), false); + + // Then: both client-facing representations carry the same sanitized, + // canonical shape. Re-serialized typed chunks are used on route paths + // that cannot forward raw bytes, so provider internals must be absent. + let Some(StreamChunk::Chat(chunk)) = &rewritten.chunk else { + panic!("expected a rewritten chat chunk"); + }; + let typed = serde_json::to_value(chunk).unwrap(); + assert!( + typed.get("prompt_token_ids").is_none(), + "typed chunk must not re-expose provider prompt_token_ids" + ); + assert!( + typed.get("prompt_sha256").is_none(), + "typed chunk must not re-expose provider prompt_sha256" + ); + assert_eq!(typed["model"], "canonical/model"); + + let raw_line = std::str::from_utf8(&rewritten.raw_bytes).unwrap(); + let raw: serde_json::Value = serde_json::from_str( + raw_line + .strip_prefix("data: ") + .expect("rewritten event is a framed SSE data line") + .trim_end(), + ) + .unwrap(); + assert_eq!(raw.get("id"), typed.get("id")); + assert!( + raw["id"].as_str().is_some_and(|id| !id.is_empty()), + "missing provider id must be replaced before emitting the event" + ); + } + + #[test] + fn missing_id_streams_use_stable_unique_synthetic_ids() { + // Given: two id-less frames from one completion and another id-less + // frame from a different completion. + let first_session = fresh_session(); + let second_session = fresh_session(); + let parse_id = |frame: &[u8], session: &ResponseSession| { + let event = inner_event(frame, session.synthetic_stream_id()) + .unwrap() + .unwrap(); + let Some(StreamChunk::Chat(chunk)) = event.chunk else { + panic!("expected a parsed chat chunk"); + }; + chunk.id + }; + + // When: every frame crosses the Chutes parser boundary. + let first_id = parse_id( + br#"{"object":"chat.completion.chunk","created":0,"model":"m","choices":[{"index":0,"delta":{"content":"a"}}]}"#, + &first_session, + ); + let next_id = parse_id( + br#"{"object":"chat.completion.chunk","created":0,"model":"m","choices":[{"index":0,"delta":{"content":"b"}}]}"#, + &first_session, + ); + let other_id = parse_id( + br#"{"object":"chat.completion.chunk","created":0,"model":"m","choices":[{"index":0,"delta":{"content":"c"}}]}"#, + &second_session, + ); + + // Then: identity is visibly gateway-synthetic, stable within one stream, + // and unique across completions. + assert!( + first_id.starts_with("chutes-gateway-"), + "missing provider id must be visibly gateway-synthetic" + ); + assert_eq!(first_id, next_id, "one stream must keep one chat id"); + assert_ne!( + first_id, other_id, + "different completions must not collide in the signature key" + ); } #[test] @@ -291,7 +468,7 @@ mod tests { let frame = br#"{"id":"x","object":"chat.completion.chunk","created":0,"model":"m","choices":[],"prompt_text":null}"#; // When: the frame is parsed at the Chutes boundary. - let event = inner_event(frame).unwrap().unwrap(); + let event = inner_event(frame, SYNTHETIC_STREAM_ID).unwrap().unwrap(); // Then: the shared chunk's flatten map preserves the provider field. let Some(StreamChunk::Chat(chunk)) = event.chunk else { @@ -309,7 +486,7 @@ mod tests { let frame = br#"{"id":"x","object":"chat.completion.chunk","created":0,"model":"m"}"#; // When: it crosses the Chutes parser boundary. - let error = inner_event(frame).unwrap_err(); + let error = inner_event(frame, SYNTHETIC_STREAM_ID).unwrap_err(); // Then: tolerance for a missing id does not weaken required completion data. assert!(format!("{error}").contains("missing field `choices`")); @@ -317,11 +494,11 @@ mod tests { #[test] fn inner_event_done_and_empty() { - assert!(inner_event(b"data: [DONE]") + assert!(inner_event(b"data: [DONE]", SYNTHETIC_STREAM_ID) .unwrap() .unwrap() .is_done_marker()); - assert!(inner_event(b" ").unwrap().is_none()); + assert!(inner_event(b" ", SYNTHETIC_STREAM_ID).unwrap().is_none()); } #[test] @@ -329,10 +506,14 @@ mod tests { // A decrypted SSE comment / keepalive line (vLLM/SGLang emit these) must // be skipped, not fed to the JSON parser — otherwise a healthy stream dies // with a parse error mid-flight. - assert!(inner_event(b": ping").unwrap().is_none()); - assert!(inner_event(b":keepalive").unwrap().is_none()); + assert!(inner_event(b": ping", SYNTHETIC_STREAM_ID) + .unwrap() + .is_none()); + assert!(inner_event(b":keepalive", SYNTHETIC_STREAM_ID) + .unwrap() + .is_none()); // Still a fatal error for genuinely non-JSON data content. - assert!(inner_event(b"data: not json").is_err()); + assert!(inner_event(b"data: not json", SYNTHETIC_STREAM_ID).is_err()); } #[tokio::test] From 85f1e22b41a92b5964ae255d80d7401bd1265e51 Mon Sep 17 00:00:00 2001 From: Lloyd Mak Date: Fri, 28 Aug 2026 06:39:04 +0000 Subject: [PATCH 3/3] Normalize a JSON-null id alongside absent and empty Review follow-up. The match treated Value::Null as a present id: None => true, Some(Value::String(id)) => id.is_empty(), Some(_) => false, // <- "id": null landed here so a frame carrying `"id": null` got no synthetic id, and the next statement - from_value:: - then failed, because ChatCompletionChunk::id is a String and cannot deserialize from null. That reproduced the exact "Chutes stream chunk parse" error this PR exists to prevent, just via a different frame shape than the one found in production. Absent, empty and null now normalize identically. Re-serialization stays conditional, so frames carrying a real id are still emitted byte-for-byte. Verified: cargo clippy -p inference_providers --all-targets -D warnings clean; cargo test -p inference_providers --lib chutes 117 passed, 0 failed. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Fm9f3maaY2wd6yHH8FEjR5 --- .../src/attested/chutes/e2ee_stream.rs | 30 +++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/crates/inference_providers/src/attested/chutes/e2ee_stream.rs b/crates/inference_providers/src/attested/chutes/e2ee_stream.rs index 50587f603..ecea6f326 100644 --- a/crates/inference_providers/src/attested/chutes/e2ee_stream.rs +++ b/crates/inference_providers/src/attested/chutes/e2ee_stream.rs @@ -92,6 +92,7 @@ fn inner_event( let needs_synthetic_id = if let Some(object) = chunk_json.as_object_mut() { let needs_synthetic_id = match object.get("id") { None => true, + Some(serde_json::Value::Null) => true, Some(serde_json::Value::String(id)) => id.is_empty(), Some(_) => false, }; @@ -373,6 +374,35 @@ mod tests { assert_eq!(chunk.id, SYNTHETIC_STREAM_ID); } + #[test] + fn inner_event_normalizes_absent_empty_and_null_ids() { + // Given: equivalent Chutes frames with every replaceable id shape. + let frames: [&[u8]; 3] = [ + br#"{"object":"chat.completion.chunk","created":0,"model":"m","choices":[]}"#, + br#"{"id":"","object":"chat.completion.chunk","created":0,"model":"m","choices":[]}"#, + br#"{"id":null,"object":"chat.completion.chunk","created":0,"model":"m","choices":[]}"#, + ]; + + // When: each frame crosses the Chutes-only parser for the same stream. + for frame in frames { + let event = inner_event(frame, SYNTHETIC_STREAM_ID).unwrap().unwrap(); + + // Then: raw and typed representations carry the same synthetic id. + let raw: serde_json::Value = serde_json::from_slice( + event + .raw_bytes + .strip_prefix(b"data: ") + .expect("normalized event is a framed SSE data line"), + ) + .unwrap(); + let Some(StreamChunk::Chat(chunk)) = event.chunk else { + panic!("expected a parsed chat chunk"); + }; + assert_eq!(raw["id"], SYNTHETIC_STREAM_ID); + assert_eq!(chunk.id, SYNTHETIC_STREAM_ID); + } + } + #[test] fn missing_id_raw_and_typed_round_trip_sanitizes_provider_fields() { // Given: an id-less provider frame carrying fields that the Chutes