feat: add SSE event normalizer module (vllm-project#49)

## Summary

Adds `events/` module to `agentic-core` — a pure parsing library that
normalizes raw SSE data lines into typed `EventFrame` structs. This is
the foundation for tool dispatch, streaming tee, and loop control
(Phases 2–4 of the core API design in PR vllm-project#44).

- **20-variant `SSEEventType` enum** covering all Responses API event
types (text, function_call, reasoning, file_search, web_search)
- **`EventPayload` enum** with typed extraction per event (no downstream
JSON access needed)
- **`normalize_sse_line(&str) → Option<EventFrame>`** — pure function,
no state, no async
- Handles both vLLM (`response.done`) and OpenAI (`response.completed`)
wire formats
- `#[non_exhaustive]` on enums for forward compatibility
- No dependency on the executor module — lands on main independently of
PR vllm-project#46

Per discussion with @maralbahari on [PR
vllm-project#46](vllm-project#46 (comment)):
this is a separate core module to avoid bloating the accumulator. Once
PR vllm-project#46 merges, a follow-up refactors the accumulator to consume
`EventFrame` instead of inline JSON parsing.

**Validated against live vLLM** (google/gemma-4-26B-A4B-it, v0.21.0) —
cassettes recorded from real streaming responses including function_call
tool use.

## Test Plan

- 33 integration tests in `tests/event_normalizer_test.rs`:
- Per-event-type parsing (text, function_call, reasoning, response
lifecycle, content_part, file/web search)
- Edge cases: `[DONE]`, empty lines, malformed JSON, unknown events,
empty deltas, unicode
- Full streaming sessions: text-only, function_call, parallel calls,
mixed text+tool
  - Real vLLM cassette replay (recorded from live gemma-4-26B-A4B-it)
  - YAML cassette-driven tests (2 cassette files)
- `cargo clippy --workspace --all-targets -- -D warnings` — clean
- `cargo fmt --check` — clean
- All 117 workspace tests pass (no regressions)

---------

Signed-off-by: Ashwin Giridharan <girida@amazon.com>

Author: ashwing

Cargo.lock

@@ -27,6 +27,7 @@ uuid = { version = "1", features = ["v7", "serde"] }
 axum.workspace = true
 criterion = { workspace = true }
 serde_yaml = "0.9"
+serde_yml = "0.0.12"
 tokio = { workspace = true, features = ["full"] }
 
 [[bench]]

crates/agentic-core/Cargo.toml

@@ -0,0 +1,5 @@
+pub mod normalize;
+pub mod types;
+
+pub use normalize::normalize_sse_line;
+pub use types::{EventFrame, EventPayload, SSEEventType};

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

@@ -0,0 +1,183 @@
+use serde_json::Value;
+
+use super::types::{EventFrame, EventPayload, SSEEventType};
+use crate::utils::common::deserialize_from_str_opt;
+
+/// Normalize a raw SSE data line into a typed [`EventFrame`].
+///
+/// Expects input in the form `data: {...}` (the `data: ` prefix is required).
+/// Returns `None` for non-data lines, empty lines, and the `data: [DONE]`
+/// sentinel.
+#[must_use]
+pub fn normalize_sse_line(line: &str) -> Option<EventFrame> {
+    let data_str = line.strip_prefix("data: ")?;
+    if data_str == "[DONE]" {
+        return None;
+    }
+
+    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);
+
+    let payload = extract_payload(event_type, &json);
+
+    Some(EventFrame {
+        event_type,
+        payload,
+        sequence_number,
+    })
+}
+
+/// 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.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.searching" => SSEEventType::WebSearchCallSearching,
+        "response.web_search_call.completed" => SSEEventType::WebSearchCallCompleted,
+        _ => 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 {
+        SSEEventType::ResponseCreated
+        | SSEEventType::ResponseInProgress
+        | SSEEventType::ResponseCompleted
+        | SSEEventType::ResponseFailed
+        | SSEEventType::ResponseIncomplete => extract_response_payload(json),
+
+        SSEEventType::OutputItemAdded => extract_output_item_added(json),
+        SSEEventType::OutputItemDone => extract_output_item_done(json),
+
+        SSEEventType::OutputTextDelta => extract_text_delta(json),
+        SSEEventType::OutputTextDone => extract_text_done(json),
+
+        SSEEventType::FunctionCallArgumentsDelta => extract_fn_call_args_delta(json),
+        SSEEventType::FunctionCallArgumentsDone => extract_fn_call_args_done(json),
+
+        SSEEventType::ReasoningSummaryTextDelta => extract_reasoning_delta(json),
+        SSEEventType::ReasoningSummaryTextDone => extract_reasoning_done(json),
+
+        SSEEventType::ContentPartAdded
+        | SSEEventType::ContentPartDone
+        | SSEEventType::FileSearchCallSearching
+        | SSEEventType::FileSearchCallCompleted
+        | SSEEventType::WebSearchCallSearching
+        | SSEEventType::WebSearchCallCompleted
+        | SSEEventType::Other => EventPayload::Raw(json.clone()),
+    }
+}
+
+fn json_str(json: &Value, key: &str) -> String {
+    json[key].as_str().unwrap_or_default().to_string()
+}
+
+fn json_str_opt(json: &Value, key: &str) -> Option<String> {
+    json[key].as_str().map(ToString::to_string)
+}
+
+fn json_u32(json: &Value, key: &str) -> u32 {
+    u32::try_from(json[key].as_u64().unwrap_or(0)).unwrap_or(u32::MAX)
+}
+
+fn extract_response_payload(json: &Value) -> EventPayload {
+    let response = &json["response"];
+    EventPayload::Response {
+        id: json_str(response, "id"),
+        status: json_str(response, "status"),
+        usage: response.get("usage").filter(|v| !v.is_null()).cloned(),
+    }
+}
+
+fn extract_output_item_added(json: &Value) -> EventPayload {
+    let item = &json["item"];
+    EventPayload::OutputItemAdded {
+        item_id: json_str(item, "id"),
+        item_type: json_str(item, "type"),
+        output_index: json_u32(json, "output_index"),
+        name: json_str_opt(item, "name"),
+        call_id: json_str_opt(item, "call_id"),
+    }
+}
+
+fn extract_output_item_done(json: &Value) -> EventPayload {
+    let item = &json["item"];
+    EventPayload::OutputItemDone {
+        item_id: json_str(item, "id"),
+        item_type: json_str(item, "type"),
+        output_index: json_u32(json, "output_index"),
+        item: item.clone(),
+    }
+}
+
+fn extract_text_delta(json: &Value) -> EventPayload {
+    EventPayload::TextDelta {
+        delta: json_str(json, "delta"),
+        item_id: json_str(json, "item_id"),
+        output_index: json_u32(json, "output_index"),
+        content_index: json_u32(json, "content_index"),
+    }
+}
+
+fn extract_text_done(json: &Value) -> EventPayload {
+    EventPayload::TextDone {
+        text: json_str(json, "text"),
+        item_id: json_str(json, "item_id"),
+        output_index: json_u32(json, "output_index"),
+    }
+}
+
+fn extract_fn_call_args_delta(json: &Value) -> EventPayload {
+    EventPayload::FunctionCallArgsDelta {
+        delta: json_str(json, "delta"),
+        call_id: json_str_opt(json, "call_id"),
+        item_id: json_str(json, "item_id"),
+        output_index: json_u32(json, "output_index"),
+    }
+}
+
+fn extract_fn_call_args_done(json: &Value) -> EventPayload {
+    EventPayload::FunctionCallArgsDone {
+        arguments: json_str(json, "arguments"),
+        call_id: json_str_opt(json, "call_id"),
+        item_id: json_str(json, "item_id"),
+        name: json_str(json, "name"),
+        output_index: json_u32(json, "output_index"),
+    }
+}
+
+fn extract_reasoning_delta(json: &Value) -> EventPayload {
+    EventPayload::ReasoningDelta {
+        delta: json_str(json, "delta"),
+        item_id: json_str(json, "item_id"),
+    }
+}
+
+fn extract_reasoning_done(json: &Value) -> EventPayload {
+    EventPayload::ReasoningDone {
+        text: json_str(json, "text"),
+        item_id: json_str(json, "item_id"),
+    }
+}

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

@@ -0,0 +1,127 @@
+use serde_json::Value;
+
+/// Classification of SSE event types from the Responses API.
+///
+/// Covers both the `OpenAI` and vLLM wire formats (e.g. `response.done` vs
+/// `response.completed`).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+#[non_exhaustive]
+pub enum SSEEventType {
+    // Response lifecycle
+    ResponseCreated,
+    ResponseInProgress,
+    ResponseCompleted,
+    ResponseFailed,
+    ResponseIncomplete,
+
+    // Output item lifecycle
+    OutputItemAdded,
+    OutputItemDone,
+
+    // Text content
+    OutputTextDelta,
+    OutputTextDone,
+    ContentPartAdded,
+    ContentPartDone,
+
+    // Function calls
+    FunctionCallArgumentsDelta,
+    FunctionCallArgumentsDone,
+
+    // Reasoning
+    ReasoningSummaryTextDelta,
+    ReasoningSummaryTextDone,
+
+    // Built-in tool calls
+    FileSearchCallSearching,
+    FileSearchCallCompleted,
+    WebSearchCallSearching,
+    WebSearchCallCompleted,
+
+    // Catch-all for unrecognized events
+    Other,
+}
+
+/// Typed payload extracted from an SSE event's JSON data.
+#[derive(Debug, Clone)]
+#[non_exhaustive]
+pub enum EventPayload {
+    /// `response.created` / `response.completed` / `response.failed` /
+    /// `response.incomplete` / `response.in_progress`
+    Response {
+        id: String,
+        status: String,
+        usage: Option<Value>,
+    },
+
+    /// `response.output_item.added`
+    OutputItemAdded {
+        item_id: String,
+        item_type: String,
+        output_index: u32,
+        name: Option<String>,
+        call_id: Option<String>,
+    },
+
+    /// `response.output_item.done`
+    OutputItemDone {
+        item_id: String,
+        item_type: String,
+        output_index: u32,
+        item: Value,
+    },
+
+    /// `response.output_text.delta`
+    TextDelta {
+        delta: String,
+        item_id: String,
+        output_index: u32,
+        content_index: u32,
+    },
+
+    /// `response.output_text.done`
+    TextDone {
+        text: String,
+        item_id: String,
+        output_index: u32,
+    },
+
+    /// `response.function_call_arguments.delta`
+    FunctionCallArgsDelta {
+        delta: String,
+        call_id: Option<String>,
+        item_id: String,
+        output_index: u32,
+    },
+
+    /// `response.function_call_arguments.done`
+    FunctionCallArgsDone {
+        arguments: String,
+        call_id: Option<String>,
+        item_id: String,
+        name: String,
+        output_index: u32,
+    },
+
+    /// `response.reasoning_summary_text.delta`
+    ReasoningDelta { delta: String, item_id: String },
+
+    /// `response.reasoning_summary_text.done`
+    ReasoningDone { text: String, item_id: String },
+
+    /// Events we classify but don't deeply parse yet.
+    Raw(Value),
+
+    /// No meaningful payload (e.g. unparseable content).
+    None,
+}
+
+/// A normalized SSE event frame — the output of [`normalize_sse_line`].
+///
+/// [`normalize_sse_line`]: crate::events::normalize::normalize_sse_line
+#[derive(Debug, Clone)]
+pub struct EventFrame {
+    pub event_type: SSEEventType,
+    pub payload: EventPayload,
+    pub sequence_number: Option<u64>,
+}

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

@@ -1,5 +1,6 @@
 pub mod config;
 pub mod error;
+pub mod events;
 pub mod executor;
 pub mod proxy;
 pub mod readiness;