refactor: unify response processing under AgentPipeline

crates/agentic-llm-d/src/handler.rs

@@ -147,8 +147,8 @@ pub async fn persist(State(state): State<BackendState>, req: Request) -> Respons
         Err(error) => return error_response(error),
     };
     let ctx = RequestContext::from(context);
-    let stored = match decode_upstream(&ctx, upstream) {
-        Ok(payload) => commit(ctx, payload, state.exec_ctx.as_ref()).await,
+    let stored = match decode_upstream(ctx, upstream).await {
+        Ok((payload, ctx)) => commit(ctx, payload, state.exec_ctx.as_ref()).await,
         Err(error) => Err(error),
     };
     match stored {

crates/agentic-llm-d/tests/split_execution_integration.rs

@@ -153,7 +153,7 @@ async fn persist(
     ctx: &ExecutionContext,
 ) -> ExecutorResult<ResponsePayload> {
     let live = RequestContext::from(unseal(&context, &signing_key())?);
-    let payload = decode_upstream(&live, upstream)?;
+    let (payload, live) = decode_upstream(live, upstream).await?;
     commit(live, payload, ctx).await
 }
 

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

@@ -1,8 +1,12 @@
 pub mod normalize;
+mod sse;
 pub mod types;
 mod validate;
 
-pub(crate) use normalize::is_data_frame;
+pub(crate) use normalize::normalize_sse_data_checked;
 pub use normalize::normalize_sse_line;
+pub use sse::{ClassifiedSseLine, SseLine};
 pub use types::{EventFrame, EventPayload, SSEEventType, SSEItemType, WireEvent};
-pub(crate) use validate::{ValidatedFrame, ensure_supported_output_item_type, output_item_identity, validate_frame};
+pub(crate) use validate::{
+    ValidatedFrame, ensure_supported_output_item_type, expected_item_type, output_item_identity, validate_frame,
+};

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

@@ -1,6 +1,7 @@
 use serde_json::Value;
 
 use super::types::{EventFrame, EventPayload, SSEEventType, SSEItemType, ShellCommandUpdate, WireEvent};
+use super::{ClassifiedSseLine, SseLine};
 use crate::types::io::OutputItem;
 use crate::utils::common::{deserialize_from_str_opt, deserialize_from_value_opt};
 
@@ -12,38 +13,47 @@ use crate::utils::common::{deserialize_from_str_opt, deserialize_from_value_opt}
 /// sentinel.
 #[must_use]
 pub fn normalize_sse_line(line: &str) -> Option<EventFrame> {
-    let data_str = line.strip_prefix("data:")?;
-    let data_str = data_str.strip_prefix(' ').unwrap_or(data_str);
-    if data_str == "[DONE]" {
+    let ClassifiedSseLine::Data(data) = SseLine::parse(line) else {
         return None;
-    }
-
-    let json: Value = deserialize_from_str_opt(data_str)?;
-    normalize_sse_value(json)
+    };
+    normalize_sse_data_checked(&data).ok().flatten()
 }
 
-/// Whether a line carries an SSE data payload that should normalize to a frame.
-pub(crate) fn is_data_frame(line: &str) -> bool {
-    line.strip_prefix("data:")
-        .map(str::trim)
-        .is_some_and(|payload| !payload.is_empty() && payload != "[DONE]")
+#[derive(Debug, thiserror::Error)]
+#[error("upstream stream has an invalid 'output_index': expected an unsigned 32-bit integer")]
+pub(crate) struct InvalidOutputIndex;
+
+/// Shares normalization with the public adapter while preserving invalid-index
+/// errors for ingestion. Malformed JSON remains a policy decision downstream.
+pub(crate) fn normalize_sse_data_checked(data: &SseLine) -> Result<Option<EventFrame>, InvalidOutputIndex> {
+    let Some(json) = deserialize_from_str_opt::<Value>(data.as_str()) else {
+        return Ok(None);
+    };
+    normalize_sse_value(json)
 }
 
 /// Normalizes an already parsed SSE payload.
-pub(crate) fn normalize_sse_value(json: Value) -> Option<EventFrame> {
+fn normalize_sse_value(json: Value) -> Result<Option<EventFrame>, InvalidOutputIndex> {
+    if let Some(index) = json.get("output_index") {
+        index
+            .as_u64()
+            .and_then(|index| u32::try_from(index).ok())
+            .ok_or(InvalidOutputIndex)?;
+    }
     let event_type = json
         .get("type")
         .and_then(Value::as_str)
         .map_or(SSEEventType::Other, SSEEventType::from);
 
     let payload = extract_payload(event_type, &json);
-    let wire: WireEvent = deserialize_from_value_opt(json)?;
-
-    Some(EventFrame {
+    let Some(wire) = deserialize_from_value_opt::<WireEvent>(json) else {
+        return Ok(None);
+    };
+    Ok(Some(EventFrame {
         event_type,
         payload,
         wire,
-    })
+    }))
 }
 
 /// Extract a typed payload from the JSON body based on the classified event type.
@@ -112,6 +122,12 @@ fn output_item_id(item: &Value) -> String {
         .to_owned()
 }
 
+fn json_output_index(json: &Value) -> Option<u32> {
+    json.get("output_index")
+        .and_then(Value::as_u64)
+        .and_then(|index| u32::try_from(index).ok())
+}
+
 fn json_u32(json: &Value, key: &str) -> u32 {
     u32::try_from(json[key].as_u64().unwrap_or(0)).unwrap_or(u32::MAX)
 }
@@ -133,7 +149,7 @@ fn extract_output_item_added(json: &Value) -> EventPayload {
     EventPayload::OutputItemAdded {
         item_id: output_item_id(item),
         item_type: SSEItemType::from(json_str(item, "type")),
-        output_index: json_u32(json, "output_index"),
+        output_index: json_output_index(json),
         name: json_str_opt(item, "name"),
         namespace: json_str_opt(item, "namespace"),
         call_id: json_str_opt(item, "call_id"),
@@ -159,7 +175,7 @@ fn extract_output_item_done(json: &Value) -> EventPayload {
     EventPayload::OutputItemDone {
         item_id,
         item_type: SSEItemType::from(json_str(&item, "type")),
-        output_index: json_u32(json, "output_index"),
+        output_index: json_output_index(json),
         item,
     }
 }
@@ -168,7 +184,7 @@ 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"),
+        output_index: json_output_index(json),
         content_index: json_u32(json, "content_index"),
     }
 }
@@ -177,7 +193,7 @@ 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"),
+        output_index: json_output_index(json),
     }
 }
 
@@ -207,7 +223,7 @@ fn extract_fn_call_args_delta(json: &Value) -> EventPayload {
         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"),
+        output_index: json_output_index(json),
     }
 }
 
@@ -217,31 +233,31 @@ fn extract_fn_call_args_done(json: &Value) -> EventPayload {
         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"),
+        output_index: json_output_index(json),
     }
 }
 
 fn extract_custom_tool_call_input_delta(json: &Value) -> EventPayload {
     EventPayload::CustomToolCallInputDelta {
         delta: json_str(json, "delta"),
         item_id: json_str(json, "item_id"),
-        output_index: json_u32(json, "output_index"),
+        output_index: json_output_index(json),
     }
 }
 
 fn extract_custom_tool_call_input_done(json: &Value) -> EventPayload {
     EventPayload::CustomToolCallInputDone {
         input: json_str(json, "input"),
         item_id: json_str(json, "item_id"),
-        output_index: json_u32(json, "output_index"),
+        output_index: json_output_index(json),
     }
 }
 
 fn extract_reasoning_text_delta(json: &Value) -> EventPayload {
     EventPayload::ReasoningTextDelta {
         delta: json_str(json, "delta"),
         item_id: json_str(json, "item_id"),
-        output_index: json_u32(json, "output_index"),
+        output_index: json_output_index(json),
         content_index: json_u32(json, "content_index"),
     }
 }
@@ -250,7 +266,7 @@ fn extract_reasoning_text_done(json: &Value) -> EventPayload {
     EventPayload::ReasoningTextDone {
         text: json_str(json, "text"),
         item_id: json_str(json, "item_id"),
-        output_index: json_u32(json, "output_index"),
+        output_index: json_output_index(json),
         content_index: json_u32(json, "content_index"),
     }
 }
@@ -259,7 +275,7 @@ fn extract_reasoning_summary_text_delta(json: &Value) -> EventPayload {
     EventPayload::ReasoningSummaryTextDelta {
         delta: json_str(json, "delta"),
         item_id: json_str(json, "item_id"),
-        output_index: json_u32(json, "output_index"),
+        output_index: json_output_index(json),
         summary_index: json_u32(json, "summary_index"),
     }
 }
@@ -268,7 +284,7 @@ fn extract_reasoning_summary_text_done(json: &Value) -> EventPayload {
     EventPayload::ReasoningSummaryTextDone {
         text: json_str(json, "text"),
         item_id: json_str(json, "item_id"),
-        output_index: json_u32(json, "output_index"),
+        output_index: json_output_index(json),
         summary_index: json_u32(json, "summary_index"),
     }
 }

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

@@ -0,0 +1,77 @@
+//! Shared SSE field classification for Responses and Messages ingestion.
+
+/// An owned SSE data payload, with the field prefix removed.
+#[derive(PartialEq, Eq)]
+pub struct SseLine(String);
+
+impl std::fmt::Debug for SseLine {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("SseLine")
+            .field("bytes", &self.0.len())
+            .finish_non_exhaustive()
+    }
+}
+
+/// Field-level classification; JSON and semantic validation happen downstream.
+#[derive(Debug, PartialEq, Eq)]
+pub enum ClassifiedSseLine {
+    Data(SseLine),
+    Done,
+    Ignore,
+}
+
+impl SseLine {
+    /// Classifies one framed line, accepting the optional space after `data:`.
+    /// Malformed JSON remains data so validation policy can decide its disposition.
+    #[must_use]
+    pub fn parse(raw: &str) -> ClassifiedSseLine {
+        let Some(data) = raw.strip_prefix("data:") else {
+            return ClassifiedSseLine::Ignore;
+        };
+        let data = data.strip_prefix(' ').unwrap_or(data);
+        match data.trim() {
+            "" => ClassifiedSseLine::Ignore,
+            "[DONE]" => ClassifiedSseLine::Done,
+            _ => ClassifiedSseLine::Data(Self(data.to_owned())),
+        }
+    }
+
+    /// Returns the data payload without trimming its contents.
+    #[must_use]
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{ClassifiedSseLine, SseLine};
+
+    #[test]
+    fn sse_line_accepts_optional_space_without_parsing_json() {
+        for raw in ["data:{", "data: {"] {
+            let ClassifiedSseLine::Data(line) = SseLine::parse(raw) else {
+                panic!("malformed JSON must remain a data payload");
+            };
+            assert_eq!(line.as_str(), "{");
+        }
+    }
+
+    #[test]
+    fn sse_line_distinguishes_done_from_ignored_fields() {
+        for raw in ["data:[DONE]", "data: [DONE]", "data: [DONE]\r\n"] {
+            assert_eq!(SseLine::parse(raw), ClassifiedSseLine::Done);
+        }
+        for raw in ["", ": heartbeat", "event: response.completed", "data:", "data: \t"] {
+            assert_eq!(SseLine::parse(raw), ClassifiedSseLine::Ignore);
+        }
+    }
+
+    #[test]
+    fn sse_line_preserves_data_whitespace() {
+        let ClassifiedSseLine::Data(line) = SseLine::parse("data:  {\"text\":\"  hello  \"} ") else {
+            panic!("expected a data payload");
+        };
+        assert_eq!(line.as_str(), " {\"text\":\"  hello  \"} ");
+    }
+}