From 3d932be75f7cd25d087db2a9db890d29ecb71a54 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 30 Aug 2026 20:52:44 +0000 Subject: [PATCH] feat(adapters): SSE non-JSON payloads carry data+event wrapper (FWD-17) Option (b) of the FWD-17 decision: payloads decode as JSON when valid (number stays number, quoted string stays string); non-JSON payloads surface as {"data": , "event": } instead of silently degrading to a JSON string. The parser captures the frame's event: field (WHATWG last-wins) and resets all pending state on every blank-line dispatch, so an event:-only frame (no data) cannot leak its name into a later frame. JSON frames surface as themselves even under a named event; the implicit 'message' default never wraps a payload. Verification: cargo test (33 forward tests incl. 4 new FWD-17 contract/parser pins), clippy, fmt. --- src/adapters/forward.rs | 153 ++++++++++++++++++++++++++++++++++++++-- 1 file changed, 148 insertions(+), 5 deletions(-) diff --git a/src/adapters/forward.rs b/src/adapters/forward.rs index d4e3677..3fbea16 100644 --- a/src/adapters/forward.rs +++ b/src/adapters/forward.rs @@ -804,13 +804,29 @@ async fn error_envelope( ResponseEnvelope::error(request_id, CallError::new(code, message, false)) } -/// Converts a parsed SSE event into a response envelope, JSON-decoding -/// the data payload when possible. +/// Maps a parsed SSE frame to a response envelope under the FWD-17 +/// payload contract: a payload that is valid JSON surfaces as the +/// decoded value itself (a number payload stays a number, a +/// quoted-string payload stays a string — the distinction a raw-text +/// fallback would erase); any other payload surfaces as the +/// `{"data": , "event": }` wrapper, so +/// a non-JSON stream is field-addressable and an upstream's +/// named-event conventions are visible on it. An empty payload is +/// `Null`. JSON payloads intentionally carry no `event` binding: a +/// JSON frame surfaces as itself (shape stability for JSON-first +/// upstreams, and the spec's implicit `event: message` default never +/// wraps a payload). fn sse_event_envelope(event: SseEvent, request_id: &str) -> ResponseEnvelope { let parsed = if event.data.trim().is_empty() { Value::Null } else { - serde_json::from_str(&event.data).unwrap_or(Value::String(event.data.clone())) + match serde_json::from_str::(&event.data) { + Ok(value) => value, + Err(_) => serde_json::json!({ + "data": event.data, + "event": event.event, + }), + } }; ResponseEnvelope::ok(request_id, parsed) } @@ -1022,9 +1038,14 @@ fn is_sse_content_type(content_type: &str) -> bool { content_type.split(';').next().unwrap_or_default().trim() == "text/event-stream" } -/// A parsed SSE event: the `data:` lines joined with `\n`. +/// A parsed SSE event: the `data:` lines joined with `\n`, plus the +/// frame's `event:` field name when the upstream sent one (WHATWG +/// event-stream semantics: the last `event:` line before the blank +/// line wins; absent → `None`, never the implicit `message` default — +/// consumers distinguish "upstream named it" from "default name"). pub(crate) struct SseEvent { pub(crate) data: String, + pub(crate) event: Option, } #[derive(Debug, thiserror::Error)] @@ -1053,13 +1074,16 @@ pub(crate) const SSE_EVENT_BUFFER_CAP: usize = 1024 * 1024; /// a chunk boundary is therefore not corrupted. Framing follows the /// WHATWG `text/event-stream` draft semantics for what the call /// protocol needs: lines split on `\n` with optional trailing `\r`; -/// `data:` fields accumulate and join with `\n`; `event:`, `id:`, and +/// `data:` fields accumulate and join with `\n`; `event:` is captured +/// per frame (last one before the blank line wins) and surfaced on +/// [`SseEvent::event`] for non-JSON payloads (FWD-17); `id:` and /// `retry:` fields are accepted and ignored; a blank line dispatches /// the pending event; a pending event with data is dispatched at EOF. pub(crate) struct SseParser { buf: Vec, data_lines: Vec, data_seen: bool, + event_name: Option, bom_stripped: bool, } @@ -1069,6 +1093,7 @@ impl SseParser { buf: Vec::new(), data_lines: Vec::new(), data_seen: false, + event_name: None, bom_stripped: false, } } @@ -1135,6 +1160,7 @@ impl SseParser { return if self.data_seen { self.complete_event() } else { + self.discard_event(); None }; } @@ -1148,6 +1174,8 @@ impl SseParser { if field == "data" { self.data_lines.push(value.to_string()); self.data_seen = true; + } else if field == "event" && !value.is_empty() { + self.event_name = Some(value.to_string()); } None } @@ -1156,8 +1184,15 @@ impl SseParser { self.data_seen = false; Some(SseEvent { data: std::mem::take(&mut self.data_lines).join("\n"), + event: self.event_name.take(), }) } + + fn discard_event(&mut self) { + self.data_seen = false; + self.event_name = None; + self.data_lines.clear(); + } } #[cfg(test)] @@ -1781,6 +1816,114 @@ mod tests { assert!(envelopes[1].result.is_ok()); } + /// FWD-17: a valid-JSON payload surfaces as the decoded value + /// itself; a non-JSON payload surfaces as the + /// `{"data", "event"}` wrapper. Numbers-vs-quoted-strings are both + /// valid JSON, so both decode (the raw-text fallback would have + /// erased that distinction). + #[tokio::test] + async fn sse_payload_contract_decodes_json_and_wraps_non_json() { + let base = spawn_responder(TestArc::new(|_parts| { + http_response( + 200, + "text/event-stream", + b"data: 123\n\ndata: \"123\"\n\ndata: plain text\n\ndata: {\"n\":1}\n\n".to_vec(), + ) + })) + .await; + let stream = forward_stream( + &minimal_client(), + &base, + "/x", + "GET", + &None, + &TestHashMap::new(), + "svc", + &serde_json::json!({"type": "object"}), + &[], + json!({}), + noop_context(), + ); + let envelopes = collect_stream(stream).await; + assert_eq!(envelopes.len(), 4); + assert_eq!(envelopes[0].result.clone().unwrap(), json!(123)); + assert_eq!(envelopes[1].result.clone().unwrap(), json!("123")); + assert_eq!( + envelopes[2].result.clone().unwrap(), + json!({"data": "plain text", "event": null}) + ); + assert_eq!(envelopes[3].result.clone().unwrap(), json!({"n": 1})); + } + + /// FWD-17: an upstream `event:` field on a non-JSON payload is + /// carried in the wrapper; JSON payloads surface as themselves even + /// under a named event, and the name is reset after dispatch (the + /// second frame must not inherit the first frame's name). + #[tokio::test] + async fn sse_event_field_carrys_on_the_non_json_wrapper_and_resets() { + let base = spawn_responder(TestArc::new(|_parts| { + http_response( + 200, + "text/event-stream", + b"event: error\ndata: upstream exploded\n\nevent: custom\ndata: {\"n\":1}\n\ndata: after\n\n" + .to_vec(), + ) + })) + .await; + let stream = forward_stream( + &minimal_client(), + &base, + "/x", + "GET", + &None, + &TestHashMap::new(), + "svc", + &serde_json::json!({"type": "object"}), + &[], + json!({}), + noop_context(), + ); + let envelopes = collect_stream(stream).await; + assert_eq!(envelopes.len(), 3); + assert_eq!( + envelopes[0].result.clone().unwrap(), + json!({"data": "upstream exploded", "event": "error"}) + ); + assert_eq!(envelopes[1].result.clone().unwrap(), json!({"n": 1})); + assert_eq!( + envelopes[2].result.clone().unwrap(), + json!({"data": "after", "event": null}), + "the name must not leak across frames" + ); + } + + /// FWD-17 parser pin: WHATWG last-wins for repeated `event:` lines + /// in one frame; an `event:`-only frame (no data) dispatches + /// nothing but must not leak its name into a later frame. + #[test] + fn sse_parser_last_event_wins_and_name_does_not_leak_across_frames() { + let mut parser = SseParser::new(); + let events = parser + .feed(b"event: a\nevent: b\ndata: x\n\n", false) + .expect("ascii only"); + assert_eq!(events.len(), 1); + assert_eq!(events[0].event.as_deref(), Some("b")); + assert_eq!(events[0].data, "x"); + + let events = parser + .feed(b"event: orphan\n\n", false) + .expect("ascii only"); + assert!(events.is_empty(), "no data field, nothing dispatched"); + + let events = parser.feed(b"data: next\n\n", false).expect("ascii only"); + assert_eq!(events.len(), 1); + assert_eq!( + events[0].event.as_deref(), + None, + "an event:-only frame must not leak its name" + ); + } + #[tokio::test] async fn bearer_credential_with_control_character_fails_loudly() { let base = spawn_responder(TestArc::new(|_parts| {