fix: normalize multi-round gateway streaming

crates/agentic-server-core/src/events/mod.rs

@@ -2,4 +2,4 @@ pub mod normalize;
 pub mod types;
 
 pub use normalize::normalize_sse_line;
-pub use types::{EventFrame, EventPayload, SSEEventType, SSEItemType};
+pub use types::{EventFrame, EventPayload, SSEEventType, SSEItemType, WireEvent};

crates/agentic-server-core/src/events/normalize.rs

@@ -1,6 +1,6 @@
 use serde_json::Value;
 
-use super::types::{EventFrame, EventPayload, SSEEventType, SSEItemType};
+use super::types::{EventFrame, EventPayload, SSEEventType, SSEItemType, WireEvent};
 use crate::utils::common::{deserialize_from_str_opt, deserialize_from_value_opt};
 
 /// Normalize a raw SSE data line into a typed [`EventFrame`].
@@ -16,58 +16,21 @@ pub fn normalize_sse_line(line: &str) -> Option<EventFrame> {
     }
 
     let json: Value = deserialize_from_str_opt(data_str)?;
-
     let event_type = json
         .get("type")
         .and_then(Value::as_str)
-        .map_or(SSEEventType::Other, classify_event_type);
-
-    let sequence_number = json.get("sequence_number").and_then(Value::as_u64);
+        .map_or(SSEEventType::Other, SSEEventType::from);
 
     let payload = extract_payload(event_type, &json);
+    let wire: WireEvent = deserialize_from_value_opt(json)?;
 
     Some(EventFrame {
         event_type,
         payload,
-        sequence_number,
+        wire,
     })
 }
 
-/// Map a wire-format event type string to our enum.
-fn classify_event_type(type_str: &str) -> SSEEventType {
-    match type_str {
-        "response.created" => SSEEventType::ResponseCreated,
-        "response.in_progress" => SSEEventType::ResponseInProgress,
-        "response.completed" | "response.done" => SSEEventType::ResponseCompleted,
-        "response.failed" => SSEEventType::ResponseFailed,
-        "response.incomplete" => SSEEventType::ResponseIncomplete,
-        "response.output_item.added" => SSEEventType::OutputItemAdded,
-        "response.output_item.done" => SSEEventType::OutputItemDone,
-        "response.output_text.delta" => SSEEventType::OutputTextDelta,
-        "response.output_text.done" => SSEEventType::OutputTextDone,
-        "response.content_part.added" => SSEEventType::ContentPartAdded,
-        "response.content_part.done" => SSEEventType::ContentPartDone,
-        "response.function_call_arguments.delta" => SSEEventType::FunctionCallArgumentsDelta,
-        "response.function_call_arguments.done" => SSEEventType::FunctionCallArgumentsDone,
-        "response.custom_tool_call_input.delta" => SSEEventType::CustomToolCallInputDelta,
-        "response.custom_tool_call_input.done" => SSEEventType::CustomToolCallInputDone,
-        "response.reasoning_text.delta" => SSEEventType::ReasoningTextDelta,
-        "response.reasoning_text.done" => SSEEventType::ReasoningTextDone,
-        "response.reasoning_part.added" => SSEEventType::ReasoningPartAdded,
-        "response.reasoning_part.done" => SSEEventType::ReasoningPartDone,
-        "response.reasoning_summary_text.delta" => SSEEventType::ReasoningSummaryTextDelta,
-        "response.reasoning_summary_text.done" => SSEEventType::ReasoningSummaryTextDone,
-        "response.file_search_call.searching" => SSEEventType::FileSearchCallSearching,
-        "response.file_search_call.completed" => SSEEventType::FileSearchCallCompleted,
-        "response.web_search_call.in_progress" => SSEEventType::WebSearchCallInProgress,
-        "response.web_search_call.searching" => SSEEventType::WebSearchCallSearching,
-        "response.web_search_call.completed" => SSEEventType::WebSearchCallCompleted,
-        "response.mcp_tool_call.in_progress" => SSEEventType::McpToolCallInProgress,
-        "response.mcp_tool_call.completed" => SSEEventType::McpToolCallCompleted,
-        _ => SSEEventType::Other,
-    }
-}
-
 /// Extract a typed payload from the JSON body based on the classified event type.
 fn extract_payload(event_type: SSEEventType, json: &Value) -> EventPayload {
     match event_type {

crates/agentic-server-core/src/events/types.rs

@@ -1,4 +1,5 @@
-use serde_json::Value;
+use serde::{Deserialize, Serialize};
+use serde_json::{Map, Value};
 
 use crate::types::io::ResponseUsage;
 
@@ -109,6 +110,104 @@ pub enum SSEEventType {
     Other,
 }
 
+impl From<&str> for SSEEventType {
+    fn from(value: &str) -> Self {
+        match value {
+            "response.created" => Self::ResponseCreated,
+            "response.in_progress" => Self::ResponseInProgress,
+            "response.completed" | "response.done" => Self::ResponseCompleted,
+            "response.failed" => Self::ResponseFailed,
+            "response.incomplete" => Self::ResponseIncomplete,
+            "response.output_item.added" => Self::OutputItemAdded,
+            "response.output_item.done" => Self::OutputItemDone,
+            "response.output_text.delta" => Self::OutputTextDelta,
+            "response.output_text.done" => Self::OutputTextDone,
+            "response.content_part.added" => Self::ContentPartAdded,
+            "response.content_part.done" => Self::ContentPartDone,
+            "response.function_call_arguments.delta" => Self::FunctionCallArgumentsDelta,
+            "response.function_call_arguments.done" => Self::FunctionCallArgumentsDone,
+            "response.custom_tool_call_input.delta" => Self::CustomToolCallInputDelta,
+            "response.custom_tool_call_input.done" => Self::CustomToolCallInputDone,
+            "response.reasoning_text.delta" => Self::ReasoningTextDelta,
+            "response.reasoning_text.done" => Self::ReasoningTextDone,
+            "response.reasoning_part.added" => Self::ReasoningPartAdded,
+            "response.reasoning_part.done" => Self::ReasoningPartDone,
+            "response.reasoning_summary_text.delta" => Self::ReasoningSummaryTextDelta,
+            "response.reasoning_summary_text.done" => Self::ReasoningSummaryTextDone,
+            "response.file_search_call.searching" => Self::FileSearchCallSearching,
+            "response.file_search_call.completed" => Self::FileSearchCallCompleted,
+            "response.web_search_call.in_progress" => Self::WebSearchCallInProgress,
+            "response.web_search_call.searching" => Self::WebSearchCallSearching,
+            "response.web_search_call.completed" => Self::WebSearchCallCompleted,
+            "response.mcp_tool_call.in_progress" => Self::McpToolCallInProgress,
+            "response.mcp_tool_call.completed" => Self::McpToolCallCompleted,
+            _ => Self::Other,
+        }
+    }
+}
+
+impl TryFrom<SSEEventType> for &'static str {
+    type Error = ();
+
+    fn try_from(value: SSEEventType) -> Result<Self, Self::Error> {
+        match value {
+            SSEEventType::ResponseCreated => Ok("response.created"),
+            SSEEventType::ResponseInProgress => Ok("response.in_progress"),
+            SSEEventType::ResponseCompleted => Ok("response.completed"),
+            SSEEventType::ResponseFailed => Ok("response.failed"),
+            SSEEventType::ResponseIncomplete => Ok("response.incomplete"),
+            SSEEventType::OutputItemAdded => Ok("response.output_item.added"),
+            SSEEventType::OutputItemDone => Ok("response.output_item.done"),
+            SSEEventType::OutputTextDelta => Ok("response.output_text.delta"),
+            SSEEventType::OutputTextDone => Ok("response.output_text.done"),
+            SSEEventType::ContentPartAdded => Ok("response.content_part.added"),
+            SSEEventType::ContentPartDone => Ok("response.content_part.done"),
+            SSEEventType::FunctionCallArgumentsDelta => Ok("response.function_call_arguments.delta"),
+            SSEEventType::FunctionCallArgumentsDone => Ok("response.function_call_arguments.done"),
+            SSEEventType::CustomToolCallInputDelta => Ok("response.custom_tool_call_input.delta"),
+            SSEEventType::CustomToolCallInputDone => Ok("response.custom_tool_call_input.done"),
+            SSEEventType::ReasoningTextDelta => Ok("response.reasoning_text.delta"),
+            SSEEventType::ReasoningTextDone => Ok("response.reasoning_text.done"),
+            SSEEventType::ReasoningPartAdded => Ok("response.reasoning_part.added"),
+            SSEEventType::ReasoningPartDone => Ok("response.reasoning_part.done"),
+            SSEEventType::ReasoningSummaryTextDelta => Ok("response.reasoning_summary_text.delta"),
+            SSEEventType::ReasoningSummaryTextDone => Ok("response.reasoning_summary_text.done"),
+            SSEEventType::FileSearchCallSearching => Ok("response.file_search_call.searching"),
+            SSEEventType::FileSearchCallCompleted => Ok("response.file_search_call.completed"),
+            SSEEventType::WebSearchCallInProgress => Ok("response.web_search_call.in_progress"),
+            SSEEventType::WebSearchCallSearching => Ok("response.web_search_call.searching"),
+            SSEEventType::WebSearchCallCompleted => Ok("response.web_search_call.completed"),
+            SSEEventType::McpToolCallInProgress => Ok("response.mcp_tool_call.in_progress"),
+            SSEEventType::McpToolCallCompleted => Ok("response.mcp_tool_call.completed"),
+            SSEEventType::Other => Err(()),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct WireEvent {
+    #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
+    pub event_type: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub sequence_number: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub output_index: Option<u64>,
+    #[serde(flatten)]
+    pub rest: Map<String, Value>,
+}
+
+impl WireEvent {
+    #[must_use]
+    pub fn new(event_type: impl Into<String>) -> Self {
+        Self {
+            event_type: Some(event_type.into()),
+            sequence_number: None,
+            output_index: None,
+            rest: Map::new(),
+        }
+    }
+}
+
 /// Typed payload extracted from an SSE event's JSON data.
 #[derive(Debug, Clone)]
 #[non_exhaustive]
@@ -205,5 +304,70 @@ pub enum EventPayload {
 pub struct EventFrame {
     pub event_type: SSEEventType,
     pub payload: EventPayload,
-    pub sequence_number: Option<u64>,
+    pub wire: WireEvent,
+}
+
+impl EventFrame {
+    #[must_use]
+    pub fn synthetic(event_type: SSEEventType, rest: Map<String, Value>) -> Option<Self> {
+        let event_type_name = <&str>::try_from(event_type).ok()?;
+        Some(Self {
+            event_type,
+            payload: EventPayload::None,
+            wire: WireEvent {
+                event_type: Some(event_type_name.to_owned()),
+                sequence_number: None,
+                output_index: None,
+                rest,
+            },
+        })
+    }
+
+    #[must_use]
+    pub fn sequence_number(&self) -> Option<u64> {
+        self.wire.sequence_number
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::SSEEventType;
+
+    #[test]
+    fn sse_event_type_wire_names_round_trip() {
+        for event_type in [
+            SSEEventType::ResponseCreated,
+            SSEEventType::ResponseInProgress,
+            SSEEventType::ResponseCompleted,
+            SSEEventType::ResponseFailed,
+            SSEEventType::ResponseIncomplete,
+            SSEEventType::OutputItemAdded,
+            SSEEventType::OutputItemDone,
+            SSEEventType::OutputTextDelta,
+            SSEEventType::OutputTextDone,
+            SSEEventType::ContentPartAdded,
+            SSEEventType::ContentPartDone,
+            SSEEventType::FunctionCallArgumentsDelta,
+            SSEEventType::FunctionCallArgumentsDone,
+            SSEEventType::CustomToolCallInputDelta,
+            SSEEventType::CustomToolCallInputDone,
+            SSEEventType::ReasoningTextDelta,
+            SSEEventType::ReasoningTextDone,
+            SSEEventType::ReasoningPartAdded,
+            SSEEventType::ReasoningPartDone,
+            SSEEventType::ReasoningSummaryTextDelta,
+            SSEEventType::ReasoningSummaryTextDone,
+            SSEEventType::FileSearchCallSearching,
+            SSEEventType::FileSearchCallCompleted,
+            SSEEventType::WebSearchCallInProgress,
+            SSEEventType::WebSearchCallSearching,
+            SSEEventType::WebSearchCallCompleted,
+            SSEEventType::McpToolCallInProgress,
+            SSEEventType::McpToolCallCompleted,
+        ] {
+            let wire_name = <&str>::try_from(event_type).expect("known event type has a wire name");
+            assert_eq!(SSEEventType::from(wire_name), event_type);
+        }
+        assert!(<&str>::try_from(SSEEventType::Other).is_err());
+    }
 }